Add possibility to authorized external access token to login (#3)

README.md

@@ -47,6 +47,43 @@ The service expects authentication service information at $ISSUER_URL/.well-know
 
 See [JSON schema](schemas/qwc-oidc-auth.json) for optional configuration options.
 
+#### Configure Access Token endpoint
+
+It is possible to authorize connection with a external Access Token in  the Authorization Header (endpoint `/tokenlogin`).
+
+For each token a configuration needs to be add in `authorized_api_token`.
+
+Example:
+```json
+{
+  "$schema": "https://github.com/qwc-services/qwc-oidc-auth/raw/main/schemas/qwc-oidc-auth.json",
+  "service": "oidc-auth",
+  "config": {
+    "issuer_url": "https://qwc2-dev.onelogin.com/oidc/2",
+    "client_id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxxxxxxx",
+    "client_secret": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+    "authorized_api_token": [{
+      "keys_url": "https://public_keys_url_to_decode_token",
+      "claims_options":{
+        "iss": {
+            "essential": true,
+            "values": ["https://example.com", "https://example.org"]
+        },
+        "sub": {
+            "essential": true,
+            "value": "xxxxxxxxxxxxx"
+        },
+        "aud": {
+          "essential": true,
+          "value": "api://xxxx-xxxxxxxxx-xxxxx"
+        }
+      }
+    }]
+  }
+}
+```
+
+`claims_options` are the token validation parameters which allow fine control over the content of the payload. See https://docs.authlib.org/en/latest/jose/jwt.html#jwt-payload-claims-validation.
 
 ### Identity provider configuration
 

schemas/qwc-oidc-auth.json

@@ -49,7 +49,24 @@
         "groupinfo": {
           "description": "Attribute name of group memberships",
           "type": "string"
-        }
+        },
+        "authorized_api_token": {
+          "description": "List of api token authorized to use tokenlogin endpoint",
+          "type": "array",
+          "items": {
+            "type": "object",
+            "properties": {
+              "keys_url": {
+                "description": "Public keys URL to decode token",
+                "type": "string"
+              },
+              "claims_options":{
+                "description": "Token validation parameters following authlib specs : https://docs.authlib.org/en/latest/jose/jwt.html#jwt-payload-claims-validation",
+                "type": "object"
+              }
+            }
+          }
+        }          
       },
       "required": [
         "issuer_url", "client_id", "client_secret"

src/server.py

@@ -1,5 +1,11 @@
 import os
 import logging
+import requests
+from authlib.oauth2.rfc6749 import TokenValidator
+from authlib.oauth2.rfc6750 import errors
+from authlib.integrations.flask_oauth2 import ResourceProtector, current_token
+from authlib.jose import jwk, jwt as auth_jwt, JWTClaims
+
 from flask import (
     Flask, url_for, jsonify, request, session, redirect, make_response
 )
@@ -78,6 +84,47 @@ def tenant_base():
     return prefix.rstrip('/') + '/'
 
 
+class APITokenValidator(TokenValidator):
+    def authenticate_token(self, token_string):
+        config = config_handler.tenant_config(tenant_handler.tenant())
+        authorized_api_token = config.get('authorized_api_token', None)
+        if authorized_api_token:
+            for api in authorized_api_token:
+                def load_key(header, payload):
+                    jwk_set = requests.get(api["keys_url"]).json()
+                    app.logger.debug(f"header = {header}")
+                    try:
+                        return jwk.loads(jwk_set, header.get('kid'))
+                    except ValueError:
+                        app.logger.debug("Invalid JSON Web Key Set")
+                        return ""
+                try:
+                    claims_options = api["claims_options"]
+                    claims_options["exp"] = {
+                        "validate": JWTClaims.validate_exp,
+                    }
+                    app.logger.debug(f"{claims_options=}")
+                    token = auth_jwt.decode(token_string, load_key, claims_options=claims_options)
+                    token.validate()
+                    token["active"] = True
+                    app.logger.debug(f"{token=}")
+                    return token
+
+                except Exception as e:
+                    app.logger.debug(f"Decode token error : {e}")
+
+        return None
+
+    def validate_token(self, token, scopes, request):
+        if not token:
+            raise errors.InvalidTokenError()
+        token.validate()
+
+
+require_oauth = ResourceProtector()
+require_oauth.register_token_validator(APITokenValidator())
+
+
 @app.route('/login')
 def login():
     config = config_handler.tenant_config(tenant_handler.tenant())
@@ -199,6 +246,43 @@ def callback():
     return resp
 
 
+@app.route('/tokenlogin')
+@require_oauth()
+def token_login():
+    userinfo = current_token
+    app.logger.info(userinfo)
+    config = config_handler.tenant_config(tenant_handler.tenant())
+    groupinfo = config.get('groupinfo', 'group')
+    mapper = GroupNameMapper()
+
+    if config.get('username'):
+        username = userinfo.get(config.get('username'))
+    else:
+        username = userinfo.get('preferred_username',
+                                userinfo.get('upn', userinfo.get('email')))
+    groups = userinfo.get(groupinfo, [])
+    if isinstance(groups, str):
+        groups = [groups]
+    # Add group for all authenticated users
+    groups.append('verified')
+    # Apply group name mappings
+    groups = [
+        mapper.mapped_group(g)
+        for g in groups
+    ]
+    identity = {'username': username, 'groups': groups}
+    app.logger.info(identity)
+    # Create the tokens we will be sending back to the user
+    access_token = create_access_token(identity)
+
+    base_url = tenant_base()
+    target_url = session.pop('target_url', base_url)
+
+    resp = make_response(redirect(target_url))
+    set_access_cookies(resp, access_token)
+    return resp
+
+
 @app.route('/logout', methods=['GET', 'POST'])
 @jwt_required(optional=True)
 def logout():