Add OpenID Connect Token authenticator

Signed-off-by: Patrick Uiterwijk <[email protected]>

Author: puiterwijk

custodia/httpd/authenticators.py

@@ -7,6 +7,8 @@
 from custodia import log
 from custodia.plugin import HTTPAuthenticator, PluginOption
 
+import requests
+
 
 class SimpleCredsAuth(HTTPAuthenticator):
     uid = PluginOption('pwd_uid', -1, "User id or name, -1 ignores user")
@@ -131,3 +133,62 @@ def handle(self, request):
         self.audit_svc_access(log.AUDIT_SVC_AUTH_FAIL,
                               request['client_id'], dn)
         return False
+
+
+class OpenIDCTokenAuth(HTTPAuthenticator):
+    token_info_url = PluginOption(str, 'token_info_url',
+                                  'URL for getting token information')
+    client_id = PluginOption(str, 'client_id',
+                             'Client ID for verifying tokens')
+    client_secret = PluginOption(str, 'client_secret',
+                                 'Client Secret for verifying tokens')
+    scope = PluginOption(str, 'scope', 'OAuth2 scope to require')
+
+    def _get_token_info(self, token):
+        return requests.post(self.token_info_url,
+                             data={'client_id': self.client_id,
+                                   'client_secret': self.client_secret,
+                                   'token': token,
+                                   'token_type_hint': 'Bearer'}).json()
+
+    def handle(self, request):
+        token = None
+        if 'Authorization' in request['headers']:
+            hdr = request['headers']['Authorization']
+            if hdr.startswith('Bearer '):
+                self.logger.debug('Bearer token provided in header')
+                token = hdr[len('Bearer '):]
+            else:
+                self.logger.debug('Unrecognized Authorization header')
+                return False
+        elif request.get('access_token'):
+            self.logger.debug('Token provided in form')
+            token = request.get('access_token')
+        else:
+            self.logger.debug('Missing any credentials in request')
+            return False
+
+        if not token:
+            self.logger.debug('No token')
+            return False
+
+        try:
+            tokeninfo = self._get_token_info(token)
+        except:
+            self.logger.debug('Error getting token information',
+                              exc_info=True)
+            return False
+
+        aud = tokeninfo['aud']
+        if isinstance(aud, list) and self.client_id not in aud:
+            self.logger.debug('My client ID not in audience list')
+            return False
+        elif self.client_id != aud:
+            self.logger.debug('Client ID is not audience')
+            return False
+
+        if self.scope not in tokeninfo['scope'].split(' '):
+            self.logger.debug('Required scope not found')
+            return False
+
+        return True

docs/source/plugins/authenticators.rst

@@ -29,3 +29,8 @@ Authenticators
     :undoc-members:
     :show-inheritance:
 
+.. autoclass:: custodia.httpd.authenticators.OpenIDCTokenAuth
+    :members:
+    :undoc-members:
+    :show-inheritance:
+

setup.py

@@ -71,6 +71,7 @@ def run(self):
     'SimpleAuthKeys = custodia.httpd.authenticators:SimpleAuthKeys',
     ('SimpleClientCertAuth = '
      'custodia.httpd.authenticators:SimpleClientCertAuth'),
+    'OIDCTokenAuth = custodia.httpd.authenticators.OpenIDCTokenAuth',
 ]
 
 custodia_authorizers = [