chore: move resources and auth classes into dedicated files

synapse_token_authenticator/config.py

@@ -9,7 +9,11 @@
     Validator,
     parse_validator,
 )
-from synapse_token_authenticator.utils import basic_auth, bearer_auth
+from synapse_token_authenticator.http_auth import (
+    HttpAuth,
+    NoAuth,
+    parse_auth,
+)
 
 
 class OIDCConfig:
@@ -104,9 +108,7 @@ class IntrospectionValidationConfig:
                 def __post_init__(self):
                     if not isinstance(self.validator, Exist):
                         self.validator = parse_validator(self.validator)
-
-                    if not isinstance(self.auth, NoAuth):
-                        self.auth = parse_auth(self.auth)
+                    self.auth = parse_auth(self.auth, context=type(self).__name__)
 
             @dataclass
             class NotifyOnRegistration:
@@ -115,8 +117,7 @@ class NotifyOnRegistration:
                 interrupt_on_error: bool = True
 
                 def __post_init__(self):
-                    if not isinstance(self.auth, NoAuth):
-                        self.auth = parse_auth(self.auth)
+                    self.auth = parse_auth(self.auth, context=type(self).__name__)
 
             @dataclass
             class OAuthConfig:
@@ -226,53 +227,3 @@ def verify_jwt_based_cfg(cfg):
     ]:
         error_msg = f"Unknown algorithm: '{cfg.algorithm}'"
         raise Exception(error_msg)
-
-
-@dataclass
-class NoAuth:
-    def header_map(self):
-        return {}
-
-
-@dataclass
-class BasicAuth:
-    username: str
-    password: str
-
-    def header_map(self):
-        return basic_auth(self.username, self.password)
-
-
-@dataclass
-class BearerAuth:
-    token: str
-
-    def header_map(self):
-        return bearer_auth(self.token)
-
-
-HttpAuth: TypeAlias = BasicAuth | BearerAuth | NoAuth
-
-
-def parse_auth(d: dict | list) -> HttpAuth:
-    if isinstance(d, dict):
-        auth_type = d.pop("type")
-        if auth_type is None:
-            return NoAuth()
-        if auth_type == "basic":
-            return BasicAuth(**d)
-        if auth_type == "bearer":
-            return BearerAuth(**d)
-        error = f"Unknown HttpAuth type {auth_type}"
-        raise Exception(error)
-    if isinstance(d, list):
-        auth_type = d.pop(0)
-        if auth_type is None:
-            return NoAuth()
-        if auth_type == "basic":
-            return BasicAuth(*d)
-        if auth_type == "bearer":
-            return BearerAuth(*d)
-        error = f"Unknown HttpAuth type {auth_type}"
-        raise Exception(error)
-    raise Exception("HttpAuth parsing failed, expected list or dict")

synapse_token_authenticator/http_auth.py

@@ -0,0 +1,90 @@
+import logging
+from base64 import b64encode
+from typing import TypeAlias
+
+from pydantic import BaseModel, ConfigDict
+
+logger = logging.getLogger(__name__)
+
+
+class NoAuth(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
+
+    def header_map(self) -> dict[bytes, list[bytes]]:
+        return {}
+
+
+class BasicAuth(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
+
+    username: str
+    password: str
+
+    def header_map(self) -> dict[bytes, list[bytes]]:
+        token = b64encode(
+            b":".join((self.username.encode("utf-8"), self.password.encode("utf-8")))
+        )
+        return {b"Authorization": [b"Basic " + token]}
+
+
+class BearerAuth(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
+
+    token: str
+
+    def header_map(self) -> dict[bytes, list[bytes]]:
+        return {b"Authorization": [b"Bearer " + self.token.encode("utf-8")]}
+
+
+HttpAuth: TypeAlias = NoAuth | BasicAuth | BearerAuth
+
+
+def parse_dict_auth(value: dict) -> HttpAuth:
+    auth_type = value.pop("type")
+    # Declaring the auth block without a 'type' parameter is an error and should raise
+    # the KeyError. If a user don't want to use authentication system, they should not
+    # include the auth block at all.
+    if auth_type is None:
+        return NoAuth()
+    if auth_type == "basic":
+        return BasicAuth(**value)
+    if auth_type == "bearer":
+        return BearerAuth(**value)
+    raise Exception(f"Unknown HttpAuth type '{auth_type}'")
+
+
+def parse_list_auth(value: list) -> HttpAuth:
+    auth_type = value.pop(0)
+    # Declaring the auth block without a 'type' information is an error and should
+    # raise the IndexError. If a user don't want to use authentication system, they
+    # should not include the auth block at all.
+    if auth_type is None:
+        return NoAuth()
+    if auth_type == "basic":
+        if len(value) != 2:
+            raise Exception("BasicAuth expects username and password")
+        return BasicAuth(username=value[0], password=value[1])
+    if auth_type == "bearer":
+        if len(value) != 1:
+            raise Exception("BearerAuth expects a single token")
+        return BearerAuth(token=value[0])
+    raise Exception(f"Unknown HttpAuth type '{auth_type}'")
+
+
+def parse_auth(
+    value: dict | list | HttpAuth, *, context: str | None = None
+) -> HttpAuth:
+    if isinstance(value, (NoAuth, BasicAuth, BearerAuth)):
+        return value
+    try:
+        if isinstance(value, dict):
+            return parse_dict_auth(value)
+        if isinstance(value, list):
+            return parse_list_auth(value)
+        raise Exception("HttpAuth parsing failed, expected list or dict")
+    except Exception as e:
+        if context:
+            logger.error("%s: HttpAuth configuration error: %s", context, e)
+        else:
+            logger.error("HttpAuth configuration error: %s", e)
+        raise e from e

synapse_token_authenticator/resources/login_metadata.py

@@ -0,0 +1,30 @@
+import json
+from urllib.parse import urljoin
+
+from twisted.web import resource
+
+from synapse_token_authenticator.config import (
+    OIDCConfig,
+)
+
+
+class LoginMetadataResource(resource.Resource):
+    def __init__(self, oidc_config: OIDCConfig):
+        self.issuer = oidc_config.issuer
+        self.metadata_url = urljoin(
+            oidc_config.issuer, "/.well-known/openid-configuration"
+        )
+        self.organization_id = oidc_config.organization_id
+        self.project_id = oidc_config.project_id
+
+    def render_GET(self, request):
+        request.setHeader(b"content-type", b"application/json")
+        request.setHeader(b"access-control-allow-origin", b"*")
+        return json.dumps(
+            {
+                "issuer": self.issuer,
+                "issuer-metadata": self.metadata_url,
+                "organization-id": self.organization_id,
+                "project-id": self.project_id,
+            }
+        ).encode("utf-8")

synapse_token_authenticator/resources/metadata.py

@@ -0,0 +1,13 @@
+import json
+
+from twisted.web import resource
+
+
+class MetadataResource(resource.Resource):
+    def __init__(self, resource: object):
+        self.resource = resource
+
+    def render_GET(self, request):
+        request.setHeader(b"content-type", b"application/json")
+        request.setHeader(b"access-control-allow-origin", b"*")
+        return json.dumps(self.resource).encode("utf-8")

synapse_token_authenticator/resources/public_key.py

@@ -0,0 +1,12 @@
+from jwcrypto.jwk import JWKSet
+from twisted.web import resource
+
+
+class PublicKeysResource(resource.Resource):
+    def __init__(self, keys: JWKSet):
+        self.keys = keys.export(private_keys=False).encode("utf-8")
+
+    def render_GET(self, request):
+        request.setHeader(b"content-type", b"application/json")
+        request.setHeader(b"access-control-allow-origin", b"*")
+        return self.keys