From 7d7e63c6e5723c25c82df93518d66cf0ebfae3a8 Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Wed, 12 Aug 2026 16:35:31 +0200 Subject: [PATCH 1/8] chore: move resources and auth classes into dedicated files --- synapse_token_authenticator/config.py | 56 ++---------- synapse_token_authenticator/http_auth.py | 85 +++++++++++++++++++ .../resources/__init__.py | 0 .../resources/login_metadata.py | 30 +++++++ .../resources/metadata.py | 13 +++ .../resources/public_key.py | 12 +++ .../token_authenticator.py | 55 +++--------- synapse_token_authenticator/utils.py | 25 ------ 8 files changed, 159 insertions(+), 117 deletions(-) create mode 100644 synapse_token_authenticator/http_auth.py create mode 100644 synapse_token_authenticator/resources/__init__.py create mode 100644 synapse_token_authenticator/resources/login_metadata.py create mode 100644 synapse_token_authenticator/resources/metadata.py create mode 100644 synapse_token_authenticator/resources/public_key.py diff --git a/synapse_token_authenticator/config.py b/synapse_token_authenticator/config.py index 59ce8bc..2b074ae 100644 --- a/synapse_token_authenticator/config.py +++ b/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: @@ -226,53 +230,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") diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py new file mode 100644 index 0000000..8a508fe --- /dev/null +++ b/synapse_token_authenticator/http_auth.py @@ -0,0 +1,85 @@ +from base64 import b64encode +from typing import Annotated, Any, TypeAlias + +from pydantic import BaseModel, BeforeValidator, ConfigDict + + +class NoAuth(BaseModel): + model_config = ConfigDict(frozen=True) + + def header_map(self) -> dict[bytes, list[bytes]]: + return {} + + +class BasicAuth(BaseModel): + model_config = ConfigDict(frozen=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) + + token: str + + def header_map(self) -> dict[bytes, list[bytes]]: + return {b"Authorization": [b"Bearer " + self.token.encode("utf-8")]} + + +def parse_auth(value: dict | list) -> NoAuth | BasicAuth | BearerAuth: + """Parse an HttpAuth config value without mutating the input.""" + if isinstance(value, dict): + data = dict(value) + try: + auth_type = data.pop("type") + except KeyError as error: + raise ValueError("HttpAuth missing type") from error + if auth_type is None: + return NoAuth() + if auth_type == "basic": + return BasicAuth(**data) + if auth_type == "bearer": + return BearerAuth(**data) + raise ValueError(f"Unknown HttpAuth type {auth_type}") + + if isinstance(value, list): + items = list(value) + if not items: + raise ValueError("HttpAuth parsing failed, empty list") + auth_type, *args = items + if auth_type is None: + return NoAuth() + if auth_type == "basic": + username, password, *rest = args + if rest: + raise ValueError("BasicAuth expects username and password") + return BasicAuth(username=username, password=password) + if auth_type == "bearer": + token, *rest = args + if rest: + raise ValueError("BearerAuth expects a single token") + return BearerAuth(token=token) + raise ValueError(f"Unknown HttpAuth type {auth_type}") + + raise ValueError("HttpAuth parsing failed, expected list or dict") + + +def _coerce_http_auth(value: Any) -> NoAuth | BasicAuth | BearerAuth: + if isinstance(value, (NoAuth, BasicAuth, BearerAuth)): + return value + if isinstance(value, (dict, list)): + return parse_auth(value) + raise ValueError("HttpAuth parsing failed, expected list or dict") + + +HttpAuth: TypeAlias = Annotated[ + NoAuth | BasicAuth | BearerAuth, + BeforeValidator(_coerce_http_auth), +] diff --git a/synapse_token_authenticator/resources/__init__.py b/synapse_token_authenticator/resources/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/synapse_token_authenticator/resources/login_metadata.py b/synapse_token_authenticator/resources/login_metadata.py new file mode 100644 index 0000000..0272ab6 --- /dev/null +++ b/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") diff --git a/synapse_token_authenticator/resources/metadata.py b/synapse_token_authenticator/resources/metadata.py new file mode 100644 index 0000000..7d5bbfb --- /dev/null +++ b/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") diff --git a/synapse_token_authenticator/resources/public_key.py b/synapse_token_authenticator/resources/public_key.py new file mode 100644 index 0000000..680c4b4 --- /dev/null +++ b/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 diff --git a/synapse_token_authenticator/token_authenticator.py b/synapse_token_authenticator/token_authenticator.py index 8940d7a..63bcf8d 100644 --- a/synapse_token_authenticator/token_authenticator.py +++ b/synapse_token_authenticator/token_authenticator.py @@ -13,15 +13,12 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import base64 -import json import logging import re from collections.abc import Awaitable, Callable from http import HTTPStatus -from typing import Any -from urllib.parse import urljoin +from typing import TYPE_CHECKING, Any -import synapse from jwcrypto import jwk, jwt from jwcrypto.common import JWException, json_decode from jwcrypto.jwk import JWKSet @@ -29,13 +26,14 @@ from synapse.module_api import ModuleApi from synapse.types import UserID from twisted.internet import defer -from twisted.web import resource -from synapse_token_authenticator.config import OIDCConfig, TokenAuthenticatorConfig +from synapse_token_authenticator.config import TokenAuthenticatorConfig +from synapse_token_authenticator.http_auth import BasicAuth +from synapse_token_authenticator.resources.login_metadata import LoginMetadataResource +from synapse_token_authenticator.resources.metadata import MetadataResource +from synapse_token_authenticator.resources.public_key import PublicKeysResource from synapse_token_authenticator.utils import ( - MetadataResource, all_list_elems_are_equal_return_the_elem, - basic_auth, get_oidp_metadata, get_path_in_dict, if_not_none, @@ -44,6 +42,9 @@ logger = logging.getLogger(__name__) +if TYPE_CHECKING: + import synapse + class TokenAuthenticator: __version__ = "0.13.1" @@ -75,7 +76,7 @@ def __init__(self, config: TokenAuthenticatorConfig, module_api: ModuleApi): self.api.register_web_resource( "/_famedly/login/com.famedly.login.token.oidc", - self.LoginMetadataResource(oidc), + LoginMetadataResource(oidc), ) if (cfg := getattr(self.config, "oauth", None)) is not None: @@ -102,43 +103,13 @@ def __init__(self, config: TokenAuthenticatorConfig, module_api: ModuleApi): keys = JWKSet() keys.add(cfg.enc_jwk) self.api.register_web_resource( - cfg.enc_jwks_endpoint, self.PublicKeysResource(keys) + cfg.enc_jwks_endpoint, PublicKeysResource(keys) ) auth_checkers[("com.famedly.login.token.epa", ("token",))] = self.check_epa self.api.register_password_auth_provider_callbacks(auth_checkers=auth_checkers) - 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") - - 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 - async def check_jwt_auth( self, username: str, login_type: str, login_dict: "synapse.module_api.JsonDict" ) -> ( @@ -257,7 +228,9 @@ async def check_oidc_auth( introspection_resp = await client.post_urlencoded_get_json( oidc_metadata.introspection_endpoint, data, - headers=basic_auth(oidc.client_id, oidc.client_secret), + headers=BasicAuth( + username=oidc.client_id, password=oidc.client_secret + ).header_map(), ) except HttpResponseException as e: if e.code == HTTPStatus.UNAUTHORIZED: diff --git a/synapse_token_authenticator/utils.py b/synapse_token_authenticator/utils.py index 4fb1511..5cd2cd6 100644 --- a/synapse_token_authenticator/utils.py +++ b/synapse_token_authenticator/utils.py @@ -1,10 +1,6 @@ -import json -from base64 import b64encode from typing import Any from urllib.parse import urljoin -from twisted.web import resource - class OpenIDProviderMetadata: """ @@ -27,17 +23,6 @@ async def get_oidp_metadata(issuer, client) -> OpenIDProviderMetadata: return OpenIDProviderMetadata(issuer, config) -def basic_auth(username: str, password: str) -> dict[bytes, list[bytes]]: - authorization = b64encode( - b":".join((username.encode("utf8"), password.encode("utf8"))) - ) - return {b"Authorization": [b"Basic " + authorization]} - - -def bearer_auth(token: str) -> dict[bytes, list[bytes]]: - return {b"Authorization": [b"Bearer " + token.encode("utf8")]} - - def if_not_none(f): return lambda x: (f(x) if x is not None else None) @@ -81,13 +66,3 @@ def validate_scopes(required_scopes: str | list[str], provided_scopes: str) -> b required_scopes = required_scopes.split() provided_scopes_list = provided_scopes.split() return all(scope in provided_scopes_list for scope in required_scopes) - - -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") From 754a582191b380ef5c50d39d1a04645e4b19e19e Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Thu, 13 Aug 2026 14:51:47 +0200 Subject: [PATCH 2/8] chore: add tests for http_auth.py --- synapse_token_authenticator/config.py | 8 -- synapse_token_authenticator/http_auth.py | 85 ++++++++++--------- .../token_authenticator.py | 45 +++------- tests/test_http_auth.py | 84 ++++++++++++++++++ tests/test_utils/__init__.py | 4 +- 5 files changed, 140 insertions(+), 86 deletions(-) create mode 100644 tests/test_http_auth.py diff --git a/synapse_token_authenticator/config.py b/synapse_token_authenticator/config.py index 2b074ae..36f268b 100644 --- a/synapse_token_authenticator/config.py +++ b/synapse_token_authenticator/config.py @@ -12,7 +12,6 @@ from synapse_token_authenticator.http_auth import ( HttpAuth, NoAuth, - parse_auth, ) @@ -109,19 +108,12 @@ 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) - @dataclass class NotifyOnRegistration: url: str auth: HttpAuth = field(default_factory=NoAuth) interrupt_on_error: bool = True - def __post_init__(self): - if not isinstance(self.auth, NoAuth): - self.auth = parse_auth(self.auth) - @dataclass class OAuthConfig: jwt_validation: JwtValidationConfig | None = None diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index 8a508fe..8fb2663 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -4,15 +4,21 @@ from pydantic import BaseModel, BeforeValidator, ConfigDict +class AuthValidationError(ValueError): + def __init__(self, message: str): + self.message = message + super().__init__(message) + + class NoAuth(BaseModel): - model_config = ConfigDict(frozen=True) + 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) + model_config = ConfigDict(frozen=True, extra="forbid", strict=True) username: str password: str @@ -25,7 +31,7 @@ def header_map(self) -> dict[bytes, list[bytes]]: class BearerAuth(BaseModel): - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, extra="forbid", strict=True) token: str @@ -33,50 +39,45 @@ def header_map(self) -> dict[bytes, list[bytes]]: return {b"Authorization": [b"Bearer " + self.token.encode("utf-8")]} -def parse_auth(value: dict | list) -> NoAuth | BasicAuth | BearerAuth: - """Parse an HttpAuth config value without mutating the input.""" - if isinstance(value, dict): - data = dict(value) - try: - auth_type = data.pop("type") - except KeyError as error: - raise ValueError("HttpAuth missing type") from error - if auth_type is None: - return NoAuth() - if auth_type == "basic": - return BasicAuth(**data) - if auth_type == "bearer": - return BearerAuth(**data) - raise ValueError(f"Unknown HttpAuth type {auth_type}") - - if isinstance(value, list): - items = list(value) - if not items: - raise ValueError("HttpAuth parsing failed, empty list") - auth_type, *args = items - if auth_type is None: - return NoAuth() - if auth_type == "basic": - username, password, *rest = args - if rest: - raise ValueError("BasicAuth expects username and password") - return BasicAuth(username=username, password=password) - if auth_type == "bearer": - token, *rest = args - if rest: - raise ValueError("BearerAuth expects a single token") - return BearerAuth(token=token) - raise ValueError(f"Unknown HttpAuth type {auth_type}") - - raise ValueError("HttpAuth parsing failed, expected list or dict") +def parse_dict_auth(value: dict) -> NoAuth | BasicAuth | BearerAuth: + try: + auth_type = value.pop("type") + except KeyError as error: + raise AuthValidationError("HttpAuth missing type") from error + if auth_type is None: + return NoAuth() + if auth_type == "basic": + return BasicAuth(username=value["username"], password=value["password"]) + if auth_type == "bearer": + return BearerAuth(token=value["token"]) + raise AuthValidationError(f"Unknown HttpAuth type {auth_type}") + + +def parse_list_auth(value: list) -> NoAuth | BasicAuth | BearerAuth: + if not value: + raise AuthValidationError("HttpAuth parsing failed, empty list") + auth_type, *args = value + if auth_type is None: + return NoAuth() + if auth_type == "basic": + if len(args) != 2: + raise AuthValidationError("BasicAuth expects username and password") + return BasicAuth(username=args[0], password=args[1]) + if auth_type == "bearer": + if len(args) != 1: + raise AuthValidationError("BearerAuth expects a single token") + return BearerAuth(token=args[0]) + raise AuthValidationError(f"Unknown HttpAuth type {auth_type}") def _coerce_http_auth(value: Any) -> NoAuth | BasicAuth | BearerAuth: if isinstance(value, (NoAuth, BasicAuth, BearerAuth)): return value - if isinstance(value, (dict, list)): - return parse_auth(value) - raise ValueError("HttpAuth parsing failed, expected list or dict") + if isinstance(value, dict): + return parse_dict_auth(value) + if isinstance(value, list): + return parse_list_auth(value) + raise AuthValidationError("HttpAuth parsing failed, expected list or dict") HttpAuth: TypeAlias = Annotated[ diff --git a/synapse_token_authenticator/token_authenticator.py b/synapse_token_authenticator/token_authenticator.py index 63bcf8d..9a421e4 100644 --- a/synapse_token_authenticator/token_authenticator.py +++ b/synapse_token_authenticator/token_authenticator.py @@ -17,13 +17,13 @@ import re from collections.abc import Awaitable, Callable from http import HTTPStatus -from typing import TYPE_CHECKING, Any +from typing import Any from jwcrypto import jwk, jwt from jwcrypto.common import JWException, json_decode from jwcrypto.jwk import JWKSet from synapse.api.errors import HttpResponseException -from synapse.module_api import ModuleApi +from synapse.module_api import JsonDict, LoginResponse, ModuleApi from synapse.types import UserID from twisted.internet import defer @@ -42,9 +42,6 @@ logger = logging.getLogger(__name__) -if TYPE_CHECKING: - import synapse - class TokenAuthenticator: __version__ = "0.13.1" @@ -111,13 +108,8 @@ def __init__(self, config: TokenAuthenticatorConfig, module_api: ModuleApi): self.api.register_password_auth_provider_callbacks(auth_checkers=auth_checkers) async def check_jwt_auth( - self, username: str, login_type: str, login_dict: "synapse.module_api.JsonDict" - ) -> ( - tuple[ - str, Callable[["synapse.module_api.LoginResponse"], Awaitable[None]] | None - ] - | None - ): + self, username: str, login_type: str, login_dict: JsonDict + ) -> tuple[str, Callable[[LoginResponse], Awaitable[None]] | None] | None: logger.info("Receiving auth request") if login_type != "com.famedly.login.token": logger.info("Wrong login type") @@ -201,13 +193,8 @@ async def check_jwt_auth( return (user_id_str, None) async def check_oidc_auth( - self, username: str, login_type: str, login_dict: "synapse.module_api.JsonDict" - ) -> ( - tuple[ - str, Callable[["synapse.module_api.LoginResponse"], Awaitable[None]] | None - ] - | None - ): + self, username: str, login_type: str, login_dict: JsonDict + ) -> tuple[str, Callable[[LoginResponse], Awaitable[None]] | None] | None: logger.info("Receiving auth request") if login_type != "com.famedly.login.token.oidc": logger.info("Wrong login type") @@ -290,13 +277,8 @@ async def check_oidc_auth( return (user_id_str, None) async def check_oauth( - self, username: str, login_type: str, login_dict: "synapse.module_api.JsonDict" - ) -> ( - tuple[ - str, Callable[["synapse.module_api.LoginResponse"], Awaitable[None]] | None - ] - | None - ): + self, username: str, login_type: str, login_dict: JsonDict + ) -> tuple[str, Callable[[LoginResponse], Awaitable[None]] | None] | None: config = self.config.oauth logger.info("Receiving auth request") if login_type != "com.famedly.login.token.oauth": @@ -591,13 +573,8 @@ def get_from_set(set_): return (fully_qualified_uid, None) async def check_epa( - self, _username: str, login_type: str, login_dict: "synapse.module_api.JsonDict" - ) -> ( - tuple[ - str, Callable[["synapse.module_api.LoginResponse"], Awaitable[None]] | None - ] - | None - ): + self, _username: str, login_type: str, login_dict: JsonDict + ) -> tuple[str, Callable[[LoginResponse], Awaitable[None]] | None] | None: config = self.config.epa logger.info("Receiving auth request") if login_type != "com.famedly.login.token.epa": @@ -721,7 +698,7 @@ def _add_user_email(self, user_id, email) -> defer.Deferred: def _get_external_id( self, fully_qualified_uid: str - ) -> "defer.Deferred[list[tuple[str, str]]]": + ) -> defer.Deferred[list[tuple[str, str]]]: return defer.ensureDeferred( self.api._store.get_external_ids_by_user(fully_qualified_uid) ) diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py new file mode 100644 index 0000000..b56b98e --- /dev/null +++ b/tests/test_http_auth.py @@ -0,0 +1,84 @@ +import pytest + +from synapse_token_authenticator.http_auth import ( + BasicAuth, + BearerAuth, + NoAuth, + _coerce_http_auth, + parse_dict_auth, + parse_list_auth, +) + + +class TestHttpAuth: + def test_coerce_http_auth_invalid_format(self): + with pytest.raises(ValueError) as e: + _coerce_http_auth("something invalid") + assert e.value.args[0] == "HttpAuth parsing failed, expected list or dict" + + def test_no_auth(self): + no_auth = NoAuth() + assert no_auth.header_map() == {} + + def test_basic_auth(self): + basic_auth = BasicAuth(username="user", password="pass") + assert basic_auth.header_map() == {b"Authorization": [b"Basic dXNlcjpwYXNz"]} + + def test_bearer_auth(self): + bearer_auth = BearerAuth(token="token") + assert bearer_auth.header_map() == {b"Authorization": [b"Bearer token"]} + + def test_parse_dict_auth(self): + assert parse_dict_auth({"type": None}) == NoAuth() + assert parse_dict_auth( + {"type": "basic", "username": "user", "password": "pass"} + ) == BasicAuth(username="user", password="pass") + assert parse_dict_auth({"type": "bearer", "token": "token"}) == BearerAuth( + token="token" + ) + + def test_parse_dict_auth_missing_type(self): + with pytest.raises(ValueError) as e: + parse_dict_auth({"username": "user", "password": "pass"}) + assert e.value.args[0] == "HttpAuth missing type" + + def test_parse_dict_auth_basic_missing_username(self): + with pytest.raises(KeyError): + parse_dict_auth({"type": "basic", "password": "pass"}) + + def test_parse_dict_auth_unknown_http_auth_type(self): + with pytest.raises(ValueError) as e: + parse_dict_auth({"type": "unknown", "token": "token"}) + assert e.value.args[0] == "Unknown HttpAuth type unknown" + + def test_parse_list_auth_basic_empty_list(self): + with pytest.raises(ValueError) as e: + parse_list_auth([]) + assert e.value.args[0] == "HttpAuth parsing failed, empty list" + + def test_parse_auth_list(self): + assert parse_list_auth([None]) == NoAuth() + assert parse_list_auth(["basic", "user", "pass"]) == BasicAuth( + username="user", password="pass" + ) + assert parse_list_auth(["bearer", "token"]) == BearerAuth(token="token") + + def test_parse_list_auth_basic_missing_username(self): + with pytest.raises(ValueError) as e: + parse_list_auth(["basic", "pass"]) + assert e.value.args[0] == "BasicAuth expects username and password" + + def test_parse_list_auth_basic_extra_fields(self): + with pytest.raises(ValueError) as e: + parse_list_auth(["basic", "user", "pass", "extra", "field"]) + assert e.value.args[0] == "BasicAuth expects username and password" + + def test_parse_list_auth_bearer_extra_fields(self): + with pytest.raises(ValueError) as e: + parse_list_auth(["bearer", "token", "extra", "field"]) + assert e.value.args[0] == "BearerAuth expects a single token" + + def test_parse_list_auth_unknown_http_auth_type(self): + with pytest.raises(ValueError) as e: + parse_list_auth(["unknown"]) + assert e.value.args[0] == "Unknown HttpAuth type unknown" diff --git a/tests/test_utils/__init__.py b/tests/test_utils/__init__.py index 96c3db2..b7d4ed4 100644 --- a/tests/test_utils/__init__.py +++ b/tests/test_utils/__init__.py @@ -20,7 +20,7 @@ import sys import warnings from collections.abc import Awaitable, Callable -from typing import TYPE_CHECKING, TypeVar +from typing import TYPE_CHECKING, Self, TypeVar import attr import zope.interface @@ -116,7 +116,7 @@ def deliverBody(self, protocol: IProtocol) -> None: protocol.connectionLost(Failure(ResponseDone())) @classmethod - def json(cls, *, code: int = 200, payload: JsonSerializable) -> "FakeResponse": + def json(cls, *, code: int = 200, payload: JsonSerializable) -> Self: headers = Headers({"Content-Type": ["application/json"]}) body = json.dumps(payload).encode("utf-8") return cls(code=code, body=body, headers=headers) From 1e33a763530810667bb1e6d7bef5d1c49a545416 Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Thu, 13 Aug 2026 15:08:31 +0200 Subject: [PATCH 3/8] chore: add coerce in dataclass config --- synapse_token_authenticator/config.py | 7 ++ synapse_token_authenticator/http_auth.py | 4 +- tests/test_http_auth.py | 82 +++++++++++++++++++++++- 3 files changed, 89 insertions(+), 4 deletions(-) diff --git a/synapse_token_authenticator/config.py b/synapse_token_authenticator/config.py index 36f268b..8a2ac74 100644 --- a/synapse_token_authenticator/config.py +++ b/synapse_token_authenticator/config.py @@ -12,6 +12,7 @@ from synapse_token_authenticator.http_auth import ( HttpAuth, NoAuth, + coerce_http_auth, ) @@ -107,6 +108,8 @@ class IntrospectionValidationConfig: def __post_init__(self): if not isinstance(self.validator, Exist): self.validator = parse_validator(self.validator) + # dataclasses does not run Pydantic's BeforeValidator + self.auth = coerce_http_auth(self.auth) @dataclass class NotifyOnRegistration: @@ -114,6 +117,10 @@ class NotifyOnRegistration: auth: HttpAuth = field(default_factory=NoAuth) interrupt_on_error: bool = True + def __post_init__(self): + # dataclasses does not run Pydantic's BeforeValidator + self.auth = coerce_http_auth(self.auth) + @dataclass class OAuthConfig: jwt_validation: JwtValidationConfig | None = None diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index 8fb2663..4ecf76f 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -70,7 +70,7 @@ def parse_list_auth(value: list) -> NoAuth | BasicAuth | BearerAuth: raise AuthValidationError(f"Unknown HttpAuth type {auth_type}") -def _coerce_http_auth(value: Any) -> NoAuth | BasicAuth | BearerAuth: +def coerce_http_auth(value: Any) -> NoAuth | BasicAuth | BearerAuth: if isinstance(value, (NoAuth, BasicAuth, BearerAuth)): return value if isinstance(value, dict): @@ -82,5 +82,5 @@ def _coerce_http_auth(value: Any) -> NoAuth | BasicAuth | BearerAuth: HttpAuth: TypeAlias = Annotated[ NoAuth | BasicAuth | BearerAuth, - BeforeValidator(_coerce_http_auth), + BeforeValidator(coerce_http_auth), ] diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py index b56b98e..c5d32d3 100644 --- a/tests/test_http_auth.py +++ b/tests/test_http_auth.py @@ -1,10 +1,11 @@ import pytest +from synapse_token_authenticator.config import TokenAuthenticatorConfig from synapse_token_authenticator.http_auth import ( BasicAuth, BearerAuth, NoAuth, - _coerce_http_auth, + coerce_http_auth, parse_dict_auth, parse_list_auth, ) @@ -13,7 +14,7 @@ class TestHttpAuth: def test_coerce_http_auth_invalid_format(self): with pytest.raises(ValueError) as e: - _coerce_http_auth("something invalid") + coerce_http_auth("something invalid") assert e.value.args[0] == "HttpAuth parsing failed, expected list or dict" def test_no_auth(self): @@ -82,3 +83,80 @@ def test_parse_list_auth_unknown_http_auth_type(self): with pytest.raises(ValueError) as e: parse_list_auth(["unknown"]) assert e.value.args[0] == "Unknown HttpAuth type unknown" + + +class TestHttpAuthConfigCoercion: + def test_introspection_auth(self): + dict_cfg = TokenAuthenticatorConfig( + { + "oauth": { + "introspection_validation": { + "endpoint": "http://idp.test/introspect", + "auth": { + "type": "basic", + "username": "user", + "password": "pass", + }, + }, + "notify_on_registration": { + "url": "http://iop.test/notify", + "auth": {"type": "bearer", "token": "token"}, + }, + } + } + ) + introspection_auth = dict_cfg.oauth.introspection_validation.auth + assert introspection_auth == BasicAuth(username="user", password="pass") + assert introspection_auth.header_map() == { + b"Authorization": [b"Basic dXNlcjpwYXNz"] + } + + notify_on_registration_auth = dict_cfg.oauth.notify_on_registration.auth + assert notify_on_registration_auth == BearerAuth(token="token") + assert notify_on_registration_auth.header_map() == { + b"Authorization": [b"Bearer token"] + } + + list_cfg = TokenAuthenticatorConfig( + { + "oauth": { + "introspection_validation": { + "endpoint": "http://idp.test/introspect", + "auth": ["bearer", "token"], + }, + "notify_on_registration": { + "url": "http://iop.test/notify", + "auth": ["basic", "user", "pass"], + }, + } + } + ) + introspection_auth = list_cfg.oauth.introspection_validation.auth + assert introspection_auth == BearerAuth(token="token") + assert introspection_auth.header_map() == {b"Authorization": [b"Bearer token"]} + + notify_on_registration_auth = list_cfg.oauth.notify_on_registration.auth + assert notify_on_registration_auth == BasicAuth( + username="user", password="pass" + ) + assert notify_on_registration_auth.header_map() == { + b"Authorization": [b"Basic dXNlcjpwYXNz"] + } + + def test_auth_defaults_to_no_auth(self): + cfg = TokenAuthenticatorConfig( + { + "oauth": { + "introspection_validation": { + "endpoint": "http://idp.test/introspect", + }, + "notify_on_registration": { + "url": "http://iop.test/notify", + }, + } + } + ) + assert cfg.oauth.introspection_validation.auth == NoAuth() + assert cfg.oauth.notify_on_registration.auth == NoAuth() + assert cfg.oauth.introspection_validation.auth.header_map() == {} + assert cfg.oauth.notify_on_registration.auth.header_map() == {} From dc7562a519a0e5dfdcb87138a81ee53d2632ce43 Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Thu, 13 Aug 2026 15:49:26 +0200 Subject: [PATCH 4/8] chore: contruct Auth with remaining mapping --- synapse_token_authenticator/http_auth.py | 4 ++-- tests/test_http_auth.py | 16 +++++++++++++++- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index 4ecf76f..d1a9cab 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -47,9 +47,9 @@ def parse_dict_auth(value: dict) -> NoAuth | BasicAuth | BearerAuth: if auth_type is None: return NoAuth() if auth_type == "basic": - return BasicAuth(username=value["username"], password=value["password"]) + return BasicAuth(**value) if auth_type == "bearer": - return BearerAuth(token=value["token"]) + return BearerAuth(**value) raise AuthValidationError(f"Unknown HttpAuth type {auth_type}") diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py index c5d32d3..716bc02 100644 --- a/tests/test_http_auth.py +++ b/tests/test_http_auth.py @@ -1,4 +1,5 @@ import pytest +from pydantic import ValidationError from synapse_token_authenticator.config import TokenAuthenticatorConfig from synapse_token_authenticator.http_auth import ( @@ -43,9 +44,22 @@ def test_parse_dict_auth_missing_type(self): parse_dict_auth({"username": "user", "password": "pass"}) assert e.value.args[0] == "HttpAuth missing type" + def test_parse_dict_auth_basic_extra_fields(self): + with pytest.raises(ValidationError) as e: + parse_dict_auth( + { + "type": "basic", + "username": "user", + "password": "pass", + "extra": "field", + } + ) + assert "Extra inputs are not permitted" in str(e) + def test_parse_dict_auth_basic_missing_username(self): - with pytest.raises(KeyError): + with pytest.raises(ValidationError) as e: parse_dict_auth({"type": "basic", "password": "pass"}) + assert "Field required" in str(e) def test_parse_dict_auth_unknown_http_auth_type(self): with pytest.raises(ValueError) as e: From 482aaff6ab5efc6e6f7360097d0a64623d4372df Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Mon, 17 Aug 2026 13:00:26 +0200 Subject: [PATCH 5/8] chore: remove coerce_http_auth --- synapse_token_authenticator/config.py | 8 ++-- synapse_token_authenticator/http_auth.py | 52 +++++++++--------------- tests/test_http_auth.py | 26 ++++++------ 3 files changed, 35 insertions(+), 51 deletions(-) diff --git a/synapse_token_authenticator/config.py b/synapse_token_authenticator/config.py index 8a2ac74..9a04c96 100644 --- a/synapse_token_authenticator/config.py +++ b/synapse_token_authenticator/config.py @@ -12,7 +12,7 @@ from synapse_token_authenticator.http_auth import ( HttpAuth, NoAuth, - coerce_http_auth, + parse_auth, ) @@ -108,8 +108,7 @@ class IntrospectionValidationConfig: def __post_init__(self): if not isinstance(self.validator, Exist): self.validator = parse_validator(self.validator) - # dataclasses does not run Pydantic's BeforeValidator - self.auth = coerce_http_auth(self.auth) + self.auth = parse_auth(self.auth) @dataclass class NotifyOnRegistration: @@ -118,8 +117,7 @@ class NotifyOnRegistration: interrupt_on_error: bool = True def __post_init__(self): - # dataclasses does not run Pydantic's BeforeValidator - self.auth = coerce_http_auth(self.auth) + self.auth = parse_auth(self.auth) @dataclass class OAuthConfig: diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index d1a9cab..f739485 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -1,13 +1,7 @@ from base64 import b64encode -from typing import Annotated, Any, TypeAlias +from typing import TypeAlias -from pydantic import BaseModel, BeforeValidator, ConfigDict - - -class AuthValidationError(ValueError): - def __init__(self, message: str): - self.message = message - super().__init__(message) +from pydantic import BaseModel, ConfigDict class NoAuth(BaseModel): @@ -39,48 +33,40 @@ def header_map(self) -> dict[bytes, list[bytes]]: return {b"Authorization": [b"Bearer " + self.token.encode("utf-8")]} -def parse_dict_auth(value: dict) -> NoAuth | BasicAuth | BearerAuth: - try: - auth_type = value.pop("type") - except KeyError as error: - raise AuthValidationError("HttpAuth missing type") from error +HttpAuth: TypeAlias = NoAuth | BasicAuth | BearerAuth + + +def parse_dict_auth(value: dict) -> HttpAuth: + auth_type = value.pop("type") if auth_type is None: return NoAuth() if auth_type == "basic": return BasicAuth(**value) if auth_type == "bearer": return BearerAuth(**value) - raise AuthValidationError(f"Unknown HttpAuth type {auth_type}") + raise Exception(f"Unknown HttpAuth type {auth_type}") -def parse_list_auth(value: list) -> NoAuth | BasicAuth | BearerAuth: - if not value: - raise AuthValidationError("HttpAuth parsing failed, empty list") - auth_type, *args = value +def parse_list_auth(value: list) -> HttpAuth: + auth_type = value.pop(0) if auth_type is None: return NoAuth() if auth_type == "basic": - if len(args) != 2: - raise AuthValidationError("BasicAuth expects username and password") - return BasicAuth(username=args[0], password=args[1]) + 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(args) != 1: - raise AuthValidationError("BearerAuth expects a single token") - return BearerAuth(token=args[0]) - raise AuthValidationError(f"Unknown HttpAuth type {auth_type}") + 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 coerce_http_auth(value: Any) -> NoAuth | BasicAuth | BearerAuth: +def parse_auth(value: dict | list | HttpAuth) -> HttpAuth: if isinstance(value, (NoAuth, BasicAuth, BearerAuth)): return value if isinstance(value, dict): return parse_dict_auth(value) if isinstance(value, list): return parse_list_auth(value) - raise AuthValidationError("HttpAuth parsing failed, expected list or dict") - - -HttpAuth: TypeAlias = Annotated[ - NoAuth | BasicAuth | BearerAuth, - BeforeValidator(coerce_http_auth), -] + raise Exception("HttpAuth parsing failed, expected list or dict") diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py index 716bc02..2cf83bf 100644 --- a/tests/test_http_auth.py +++ b/tests/test_http_auth.py @@ -6,16 +6,16 @@ BasicAuth, BearerAuth, NoAuth, - coerce_http_auth, + parse_auth, parse_dict_auth, parse_list_auth, ) class TestHttpAuth: - def test_coerce_http_auth_invalid_format(self): - with pytest.raises(ValueError) as e: - coerce_http_auth("something invalid") + def test_parse_auth_invalid_format(self): + with pytest.raises(Exception) as e: + parse_auth("something invalid") assert e.value.args[0] == "HttpAuth parsing failed, expected list or dict" def test_no_auth(self): @@ -40,9 +40,9 @@ def test_parse_dict_auth(self): ) def test_parse_dict_auth_missing_type(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_dict_auth({"username": "user", "password": "pass"}) - assert e.value.args[0] == "HttpAuth missing type" + assert e.value.args[0] == "type" def test_parse_dict_auth_basic_extra_fields(self): with pytest.raises(ValidationError) as e: @@ -62,14 +62,14 @@ def test_parse_dict_auth_basic_missing_username(self): assert "Field required" in str(e) def test_parse_dict_auth_unknown_http_auth_type(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_dict_auth({"type": "unknown", "token": "token"}) assert e.value.args[0] == "Unknown HttpAuth type unknown" def test_parse_list_auth_basic_empty_list(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_list_auth([]) - assert e.value.args[0] == "HttpAuth parsing failed, empty list" + assert e.value.args[0] == "pop from empty list" def test_parse_auth_list(self): assert parse_list_auth([None]) == NoAuth() @@ -79,22 +79,22 @@ def test_parse_auth_list(self): assert parse_list_auth(["bearer", "token"]) == BearerAuth(token="token") def test_parse_list_auth_basic_missing_username(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_list_auth(["basic", "pass"]) assert e.value.args[0] == "BasicAuth expects username and password" def test_parse_list_auth_basic_extra_fields(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_list_auth(["basic", "user", "pass", "extra", "field"]) assert e.value.args[0] == "BasicAuth expects username and password" def test_parse_list_auth_bearer_extra_fields(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_list_auth(["bearer", "token", "extra", "field"]) assert e.value.args[0] == "BearerAuth expects a single token" def test_parse_list_auth_unknown_http_auth_type(self): - with pytest.raises(ValueError) as e: + with pytest.raises(Exception) as e: parse_list_auth(["unknown"]) assert e.value.args[0] == "Unknown HttpAuth type unknown" From addc0420f7d33f949dc121e052da6a9d176254dc Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Mon, 17 Aug 2026 13:51:53 +0200 Subject: [PATCH 6/8] chore: add logging context for auth config --- synapse_token_authenticator/config.py | 4 +- synapse_token_authenticator/http_auth.py | 28 ++++++++++---- tests/test_http_auth.py | 49 +++++++++++++++++++++++- 3 files changed, 69 insertions(+), 12 deletions(-) diff --git a/synapse_token_authenticator/config.py b/synapse_token_authenticator/config.py index 9a04c96..9c9cfe9 100644 --- a/synapse_token_authenticator/config.py +++ b/synapse_token_authenticator/config.py @@ -108,7 +108,7 @@ class IntrospectionValidationConfig: def __post_init__(self): if not isinstance(self.validator, Exist): self.validator = parse_validator(self.validator) - self.auth = parse_auth(self.auth) + self.auth = parse_auth(self.auth, context=type(self).__name__) @dataclass class NotifyOnRegistration: @@ -117,7 +117,7 @@ class NotifyOnRegistration: interrupt_on_error: bool = True def __post_init__(self): - self.auth = parse_auth(self.auth) + self.auth = parse_auth(self.auth, context=type(self).__name__) @dataclass class OAuthConfig: diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index f739485..04051ae 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -1,8 +1,11 @@ +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) @@ -44,7 +47,7 @@ def parse_dict_auth(value: dict) -> HttpAuth: return BasicAuth(**value) if auth_type == "bearer": return BearerAuth(**value) - raise Exception(f"Unknown HttpAuth type {auth_type}") + raise Exception(f"Unknown HttpAuth type '{auth_type}'") def parse_list_auth(value: list) -> HttpAuth: @@ -59,14 +62,23 @@ def parse_list_auth(value: list) -> HttpAuth: if len(value) != 1: raise Exception("BearerAuth expects a single token") return BearerAuth(token=value[0]) - raise Exception(f"Unknown HttpAuth type {auth_type}") + raise Exception(f"Unknown HttpAuth type '{auth_type}'") -def parse_auth(value: dict | list | HttpAuth) -> HttpAuth: +def parse_auth( + value: dict | list | HttpAuth, *, context: str | None = None +) -> HttpAuth: if isinstance(value, (NoAuth, BasicAuth, BearerAuth)): return value - 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") + 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 diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py index 2cf83bf..72c164f 100644 --- a/tests/test_http_auth.py +++ b/tests/test_http_auth.py @@ -1,3 +1,5 @@ +import logging + import pytest from pydantic import ValidationError @@ -64,7 +66,7 @@ def test_parse_dict_auth_basic_missing_username(self): def test_parse_dict_auth_unknown_http_auth_type(self): with pytest.raises(Exception) as e: parse_dict_auth({"type": "unknown", "token": "token"}) - assert e.value.args[0] == "Unknown HttpAuth type unknown" + assert e.value.args[0] == "Unknown HttpAuth type 'unknown'" def test_parse_list_auth_basic_empty_list(self): with pytest.raises(Exception) as e: @@ -96,7 +98,7 @@ def test_parse_list_auth_bearer_extra_fields(self): def test_parse_list_auth_unknown_http_auth_type(self): with pytest.raises(Exception) as e: parse_list_auth(["unknown"]) - assert e.value.args[0] == "Unknown HttpAuth type unknown" + assert e.value.args[0] == "Unknown HttpAuth type 'unknown'" class TestHttpAuthConfigCoercion: @@ -174,3 +176,46 @@ def test_auth_defaults_to_no_auth(self): assert cfg.oauth.notify_on_registration.auth == NoAuth() assert cfg.oauth.introspection_validation.auth.header_map() == {} assert cfg.oauth.notify_on_registration.auth.header_map() == {} + + def test_introspection_auth_error_logs_config_class(self, caplog): + with ( + caplog.at_level(logging.ERROR), + pytest.raises(Exception), + ): + TokenAuthenticatorConfig( + { + "oauth": { + "introspection_validation": { + "endpoint": "http://idp.test/introspect", + "auth": {"type": "unknown"}, + }, + } + } + ) + assert ( + "IntrospectionValidationConfig: HttpAuth configuration error: Unknown HttpAuth type 'unknown'" + in caplog.text + ) + + def test_notify_on_registration_auth_error_logs_config_class(self, caplog): + with ( + caplog.at_level(logging.ERROR), + pytest.raises(Exception), + ): + TokenAuthenticatorConfig( + { + "oauth": { + "introspection_validation": { + "endpoint": "http://idp.test/introspect", + }, + "notify_on_registration": { + "url": "http://iop.test/notify", + "auth": ["unknown"], + }, + } + } + ) + assert ( + "NotifyOnRegistration: HttpAuth configuration error: Unknown HttpAuth type 'unknown'" + in caplog.text + ) From c2832bfb4d695abba133122261f1dad082540fc2 Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Tue, 18 Aug 2026 12:05:17 +0200 Subject: [PATCH 7/8] chore: add comment about keyError and indexError --- synapse_token_authenticator/http_auth.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index 04051ae..d5f3680 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -41,6 +41,7 @@ def header_map(self) -> dict[bytes, list[bytes]]: def parse_dict_auth(value: dict) -> HttpAuth: auth_type = value.pop("type") + # This is not KeyError safe, but it is caught in the caller if auth_type is None: return NoAuth() if auth_type == "basic": @@ -52,6 +53,7 @@ def parse_dict_auth(value: dict) -> HttpAuth: def parse_list_auth(value: list) -> HttpAuth: auth_type = value.pop(0) + # This is not IndexError safe, but it is caught in the caller if auth_type is None: return NoAuth() if auth_type == "basic": From edaac4e420c851e459f9e12fe5ae3983e31b36c9 Mon Sep 17 00:00:00 2001 From: Soyoung Kim Date: Tue, 18 Aug 2026 13:40:10 +0200 Subject: [PATCH 8/8] chore: update comment --- synapse_token_authenticator/http_auth.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/synapse_token_authenticator/http_auth.py b/synapse_token_authenticator/http_auth.py index d5f3680..48ed132 100644 --- a/synapse_token_authenticator/http_auth.py +++ b/synapse_token_authenticator/http_auth.py @@ -41,7 +41,9 @@ def header_map(self) -> dict[bytes, list[bytes]]: def parse_dict_auth(value: dict) -> HttpAuth: auth_type = value.pop("type") - # This is not KeyError safe, but it is caught in the caller + # 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": @@ -53,7 +55,9 @@ def parse_dict_auth(value: dict) -> HttpAuth: def parse_list_auth(value: list) -> HttpAuth: auth_type = value.pop(0) - # This is not IndexError safe, but it is caught in the caller + # 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":