Skip to content

Instantly share code, notes, and snippets.

@jamielennox
Created November 3, 2016 01:19
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jamielennox/8749f39fc53a7d53c822ba76fd5271b2 to your computer and use it in GitHub Desktop.
Save jamielennox/8749f39fc53a7d53c822ba76fd5271b2 to your computer and use it in GitHub Desktop.
Test out using ServiceTokenAuthWrapper.
# there's no particular reason i used nova creds for auth_token, they just exist already
# auth_section simply points one section to the other rather than repeat it
[keystone_authtoken]
auth_uri = http://localhost/identity
auth_section = service_auth
[service_auth]
auth_type = password
auth_url = http://localhost/identity
password = password
username = nova
user_domain_name = Default
project_name = service
project_domain_name = Default
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import logging
import os
import wsgiref.simple_server
from glanceclient import client as glance_client
from keystoneauth1.access import service_catalog
from keystoneauth1 import loading
from keystoneauth1 import plugin
from keystoneauth1 import service_token
from keystonemiddleware import auth_token
from oslo_config import cfg
from oslo_context import context
import webob.dec
logging.basicConfig(level=logging.DEBUG)
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
CONF.register_opts([cfg.StrOpt('host', default='', help='Listen Host'),
cfg.IntOpt('port', default=8081, help='Listen Port')])
service_group = 'service_auth'
loading.register_auth_conf_options(CONF, group=service_group)
loading.register_session_conf_options(CONF, group=service_group)
script_dir = os.path.dirname(os.path.realpath(__file__))
default_config_files = [os.path.join(script_dir, 'service_token_test.conf')]
CONF(project='service_token_test',
version='1.0',
default_config_files=default_config_files)
# The service auth plugin that is refreshable for X-Service-Token. This is
# global because we can reuse the same one across multiple requests.
service_auth = loading.load_auth_from_conf_options(CONF, group=service_group)
assert service_auth
class ContextAuthPlugin(plugin.BaseAuthPlugin):
def __init__(self, auth_token, sc, user_id, project_id):
self.auth_token = auth_token
self.service_catalog = sc
self.user_id = user_id
self.project_id = project_id
def get_token(self, *args, **kwargs):
return self.auth_token
def get_endpoint(self, session, **kwargs):
url_args = {}
for key in ['service_type',
'interface',
'region_name',
'service_name']:
try:
url_args[key] = kwargs[key]
except KeyError:
pass
return self.service_catalog.url_for(**url_args)
def get_user_id(self, *args, **kwargs):
return self.user_id
def get_project_id(self, *args, **kwargs):
return self.project_id
@classmethod
def create_from_req(cls, req):
"""This is the problem function.
We need to take the information we have available after context has
been serialized over RPC and convert that into a usable plugin. I've
been working toward this for a while but don't know where it should
live. It's only recently that context is standardized enough to even
consider it.
"""
ctxt = context.RequestContext.from_environ(req.environ)
# nova already has some logic about creating an auth plugin from the
# context but we need to improve this and figure out a way to
# generalize it.
# https://github.com/openstack/nova/blob/5fe5185443052980ae31231f572efb17fcce050a/nova/context.py#L39
sc_dict = json.loads(req.environ.get('HTTP_X_SERVICE_CATALOG', ''))
return ContextAuthPlugin(auth_token=ctxt.auth_token,
sc=service_catalog.ServiceCatalogV2(sc_dict),
user_id=ctxt.user,
project_id=ctxt.tenant)
@webob.dec.wsgify
def application(req):
LOG.info("request received")
# extract the user auth from the user info and make it a plugin
user_auth = ContextAuthPlugin.create_from_req(req)
# Combine the user and service authentication mechansims
wrapper = service_token.ServiceTokenAuthWrapper(user_auth=user_auth,
service_auth=service_auth)
# create a session, everything using that session is service authed
session = loading.load_session_from_conf_options(CONF,
group=service_group,
auth=wrapper)
# for an example combined call we'll fetch a glance image list
gc = glance_client.Client(2, session=session)
names = [image.name for image in gc.images.list(limit=5)]
# return it so we can see it worked
return webob.Response(content_type='application/json',
status=200,
body=json.dumps(names))
app = auth_token.AuthProtocol(application, {})
server = wsgiref.simple_server.make_server(CONF.host, CONF.port, app)
LOG.info("Starting Server on '%s:%s'", CONF.host, CONF.port)
server.serve_forever()
@jamielennox
Copy link
Author

It's configured to work with my devstack output. I run:

In one terminal run:

python service_token_test.py

In another do:

export OS_CLOUD=devstack-admin
os-http get http://localhost:8081/

If you don't have os-http you can pip install it:

pip install os-http

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment