Source code for keystone.oauth1.validator

# Copyright 2014 OpenStack Foundation
# 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
# 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.

"""oAuthlib request validator."""

import six

from keystone.common import provider_api
from keystone import exception
from keystone.oauth1.backends import base
from keystone.oauth1 import core as oauth1

METHOD_NAME = 'oauth_validator'
PROVIDERS = provider_api.ProviderAPIs

[docs]class OAuthValidator(provider_api.ProviderAPIMixin, oauth1.RequestValidator): # TODO(mhu) set as option probably? @property def enforce_ssl(self): return False @property def safe_characters(self): # oauth tokens are generated from a uuid hex value return set("abcdef0123456789") def _check_token(self, token): # generic token verification when they're obtained from a uuid hex return (set(token) <= self.safe_characters and len(token) == 32)
[docs] def check_client_key(self, client_key): return self._check_token(client_key)
[docs] def check_request_token(self, request_token): return self._check_token(request_token)
[docs] def check_access_token(self, access_token): return self._check_token(access_token)
[docs] def check_nonce(self, nonce): # Assuming length is not a concern return set(nonce) <= self.safe_characters
[docs] def check_verifier(self, verifier): return (all(i in base.VERIFIER_CHARS for i in verifier) and len(verifier) == 8)
[docs] def get_client_secret(self, client_key, request): client = PROVIDERS.oauth_api.get_consumer_with_secret(client_key) return client['secret']
[docs] def get_request_token_secret(self, client_key, token, request): token_ref = PROVIDERS.oauth_api.get_request_token(token) return token_ref['request_secret']
[docs] def get_access_token_secret(self, client_key, token, request): access_token = PROVIDERS.oauth_api.get_access_token(token) return access_token['access_secret']
[docs] def get_default_realms(self, client_key, request): # realms weren't implemented with the previous library return []
[docs] def get_realms(self, token, request): return []
[docs] def get_redirect_uri(self, token, request): # OOB (out of band) is supposed to be the default value to use return 'oob'
[docs] def get_rsa_key(self, client_key, request): # HMAC signing is used, so return a dummy value return ''
[docs] def invalidate_request_token(self, client_key, request_token, request): """Invalidate a used request token. :param client_key: The client/consumer key. :param request_token: The request token string. :param request: An oauthlib.common.Request object. :returns: None Per `Section 2.3`_ of the spec: "The server MUST (...) ensure that the temporary credentials have not expired or been used before." .. _`Section 2.3`: This method should ensure that provided token won't validate anymore. It can be simply removing RequestToken from storage or setting specific flag that makes it invalid (note that such flag should be also validated during request token validation). This method is used by * AccessTokenEndpoint """ # FIXME(lbragstad): Remove the above documentation string once # is resolved. It is # being duplicated here to work around oauthlib compatibility issues # with Sphinx 1.7.5, which have been reported upstream in # # this method is invoked when an access token is generated out of a # request token, to make sure that request token cannot be consumed # anymore. This is done in the backend, so we do nothing here. pass
[docs] def validate_client_key(self, client_key, request): try: return PROVIDERS.oauth_api.get_consumer(client_key) is not None except exception.NotFound: return False
[docs] def validate_request_token(self, client_key, token, request): try: req_token = PROVIDERS.oauth_api.get_request_token(token) if req_token: return req_token['consumer_id'] == client_key else: return False except exception.NotFound: return False
[docs] def validate_access_token(self, client_key, token, request): try: return PROVIDERS.oauth_api.get_access_token(token) is not None except exception.NotFound: return False
[docs] def validate_timestamp_and_nonce(self, client_key, timestamp, nonce, request, request_token=None, access_token=None): return True
[docs] def validate_redirect_uri(self, client_key, redirect_uri, request): # we expect OOB, we don't really care return True
[docs] def validate_requested_realms(self, client_key, realms, request): # realms are not used return True
[docs] def validate_realms(self, client_key, token, request, uri=None, realms=None): return True
[docs] def validate_verifier(self, client_key, token, verifier, request): try: req_token = PROVIDERS.oauth_api.get_request_token(token) return req_token['verifier'] == verifier except exception.NotFound: return False
[docs] def verify_request_token(self, token, request): # there aren't strong expectations on the request token format return isinstance(token, six.string_types)
[docs] def verify_realms(self, token, realms, request): return True
# The following save_XXX methods are called to create tokens. I chose to # keep the original logic, but the comments below show how that could be # implemented. The real implementation logic is in the backend.
[docs] def save_access_token(self, token, request): pass
# token_duration = CONF.oauth1.request_token_duration # request_token_id = request.client_key # self.oauth_api.create_access_token(request_token_id, # token_duration, # token["oauth_token"], # token["oauth_token_secret"])
[docs] def save_request_token(self, token, request): pass
# project_id = request.headers.get('Requested-Project-Id') # token_duration = CONF.oauth1.request_token_duration # self.oauth_api.create_request_token(request.client_key, # project_id, # token_duration, # token["oauth_token"], # token["oauth_token_secret"])
[docs] def save_verifier(self, token, verifier, request): """Associate an authorization verifier with a request token. :param token: A request token string. :param verifier: A dictionary containing the oauth_verifier and oauth_token :param request: An oauthlib.common.Request object. We need to associate verifiers with tokens for validation during the access token request. Note that unlike save_x_token token here is the ``oauth_token`` token string from the request token saved previously. This method is used by * AuthorizationEndpoint """ # FIXME(lbragstad): Remove the above documentation string once # is resolved. It is # being duplicated here to work around oauthlib compatibility issues # with Sphinx 1.7.5, which have been reported upstream in # # keep the old logic for this, as it is done in two steps and requires # information that the request validator has no access to pass