diff --git a/synapse_token_authenticator/claims_validator.py b/synapse_token_authenticator/claims_validator.py
index 09179ba..ef89a6b 100644
--- a/synapse_token_authenticator/claims_validator.py
+++ b/synapse_token_authenticator/claims_validator.py
@@ -163,12 +163,18 @@ def parse_validator(d: dict | list) -> Validator:
val_type = d.pop("type")
validator = VALIDATORS.get(val_type)
if validator:
- return validator(**d)
+ try:
+ return validator(**d)
+ except TypeError as e:
+ raise InvalidClaimsValidatorError(f"Invalid validator arguments: {e}")
raise InvalidClaimsValidatorError(f"Unknown validator type {val_type}")
if isinstance(d, list):
val_type = d.pop(0)
validator = VALIDATORS.get(val_type)
if validator:
- return validator(*d)
+ try:
+ return validator(*d)
+ except TypeError as e:
+ raise InvalidClaimsValidatorError(f"Invalid validator arguments: {e}")
raise InvalidClaimsValidatorError(f"Unknown validator type {val_type}")
raise InvalidClaimsValidatorError("Validator parsing failed, expected list or dict")
diff --git a/synapse_token_authenticator/config/epa.py b/synapse_token_authenticator/config/epa.py
index 98e07b2..ac2bea8 100644
--- a/synapse_token_authenticator/config/epa.py
+++ b/synapse_token_authenticator/config/epa.py
@@ -62,6 +62,8 @@ def parse_enc_jwk(cls, value: Any) -> JWK | None:
return None
if isinstance(value, JWK):
return value
+ if isinstance(value, str):
+ return JWK.from_json(value)
if isinstance(value, dict):
return JWK(**value)
return None
@@ -74,11 +76,15 @@ def parse_jwk_set(cls, value: Any) -> JWKSet | JWK | None:
if isinstance(value, (JWKSet, JWK)):
return value
if isinstance(value, str):
- return JWKSet.from_json(value)
- if isinstance(value, dict) and "keys" in value:
- return JWKSet.from_json(json.dumps(value))
+ if json.loads(value).get("keys"):
+ return JWKSet.from_json(value)
+ else:
+ return JWK.from_json(value)
if isinstance(value, dict):
- return JWK(**value)
+ if "keys" in value:
+ return JWKSet.from_json(json.dumps(value))
+ else:
+ return JWK(**value)
return None
@model_validator(mode="after")
@@ -87,15 +93,17 @@ def decide_enc_jwk(self) -> Self:
self.enc_jwk is not None,
self.enc_jwk_file is not None,
]
- if sum(sources) != 1:
- raise ValueError("Exactly one of enc_jwk or enc_jwk_file must be set")
- if self.enc_jwk:
- return self
- elif self.enc_jwk_file:
- with open(self.enc_jwk_file, "rb") as f:
- self.enc_jwk = JWK.from_pem(f.read())
+ if sum(sources) == 1:
+ if self.enc_jwk:
return self
- raise ValueError("No encryption JWK")
+ elif self.enc_jwk_file:
+ try:
+ with open(self.enc_jwk_file, "rb") as f:
+ self.enc_jwk = JWK.from_pem(f.read())
+ return self
+ except FileNotFoundError:
+ raise ValueError(f"enc_jwk file '{self.enc_jwk_file}' not found")
+ raise ValueError("Exactly one of enc_jwk or enc_jwk_file must be set")
@model_validator(mode="after")
def decide_jwk_set(self) -> Self:
@@ -104,16 +112,18 @@ def decide_jwk_set(self) -> Self:
self.jwk_file is not None,
self.jwks_endpoint is not None,
]
- if sum(sources) != 1:
- raise ValueError(
- "Exactly one of jwk_set, jwk_file, or jwks_endpoint must be set"
- )
- if self.jwk_set:
- return self
- elif self.jwk_file:
- with open(self.jwk_file, "rb") as f:
- self.jwk_set = JWK.from_pem(f.read())
+ if sum(sources) == 1:
+ if self.jwk_set:
+ return self
+ elif self.jwk_file:
+ try:
+ with open(self.jwk_file, "rb") as f:
+ self.jwk_set = JWK.from_pem(f.read())
+ return self
+ except FileNotFoundError:
+ raise ValueError(f"jwk_file '{self.jwk_file}' not found")
+ elif self.jwks_endpoint:
return self
- elif self.jwks_endpoint:
- return self
- raise ValueError("No JWK set")
+ raise ValueError(
+ "Exactly one of jwk_set, jwk_file, or jwks_endpoint must be set"
+ )
diff --git a/synapse_token_authenticator/config/oauth.py b/synapse_token_authenticator/config/oauth.py
index 21cd7c2..24caed5 100644
--- a/synapse_token_authenticator/config/oauth.py
+++ b/synapse_token_authenticator/config/oauth.py
@@ -56,11 +56,15 @@ def parse_jwk_set(cls, value: Any) -> JWKSet | JWK | None:
if isinstance(value, (JWKSet, JWK)):
return value
if isinstance(value, str):
- return JWKSet.from_json(value)
- if isinstance(value, dict) and "keys" in value:
- return JWKSet.from_json(json.dumps(value))
+ if json.loads(value).get("keys"):
+ return JWKSet.from_json(value)
+ else:
+ return JWK.from_json(value)
if isinstance(value, dict):
- return JWK(**value)
+ if "keys" in value:
+ return JWKSet.from_json(json.dumps(value))
+ else:
+ return JWK(**value)
return None
@model_validator(mode="after")
@@ -70,19 +74,21 @@ def decide_jwk_set(self) -> Self:
self.jwk_file is not None,
self.jwks_endpoint is not None,
]
- if sum(sources) != 1:
- raise ValueError(
- "Exactly one of jwk_set, jwk_file, or jwks_endpoint must be set"
- )
- if self.jwk_set:
- return self
- elif self.jwk_file:
- with open(self.jwk_file, "rb") as f:
- self.jwk_set = JWK.from_pem(f.read())
+ if sum(sources) == 1:
+ if self.jwk_set:
+ return self
+ elif self.jwk_file:
+ try:
+ with open(self.jwk_file, "rb") as f:
+ self.jwk_set = JWK.from_pem(f.read())
+ return self
+ except FileNotFoundError:
+ raise ValueError(f"jwk_file '{self.jwk_file}' not found")
+ elif self.jwks_endpoint:
return self
- elif self.jwks_endpoint:
- return self
- raise ValueError("No JWK set")
+ raise ValueError(
+ "Exactly one of jwk_set, jwk_file, or jwks_endpoint must be set"
+ )
@dataclass(config=ConfigDict(arbitrary_types_allowed=True, extra="ignore"))
diff --git a/synapse_token_authenticator/token_authenticator.py b/synapse_token_authenticator/token_authenticator.py
index 146d669..7496936 100644
--- a/synapse_token_authenticator/token_authenticator.py
+++ b/synapse_token_authenticator/token_authenticator.py
@@ -101,6 +101,10 @@ def __init__(self, config: TokenAuthenticatorConfig, module_api: ModuleApi):
# Registers the encryption public keys
keys = JWKSet()
+
+ # enc_jwk and enc_jwk_file are both optional fields but either one must set.
+ # If enc_jwk is empty, the model resolves it from enc_jwk_file. So enc_jwk
+ # cannot be empty. This assert is to resolve mypy error.
assert self.config.epa.enc_jwk is not None
keys.add(self.config.epa.enc_jwk)
self.api.register_web_resource(
diff --git a/tests/__init__.py b/tests/__init__.py
index 880f786..e1655a6 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -178,10 +178,12 @@ def get_jwt_token(
claims=None,
id_="123456",
extra_headers=None,
+ key=None,
) -> str:
if extra_headers is None:
extra_headers = {}
- key = get_jwk(secret, id_)
+ if key is None:
+ key = get_jwk(secret, id_)
if claims is None:
claims = {}
claims["sub"] = username
diff --git a/tests/test_config/test_epa.py b/tests/test_config/test_epa.py
index 405fb6f..c6fdbc0 100644
--- a/tests/test_config/test_epa.py
+++ b/tests/test_config/test_epa.py
@@ -2,20 +2,18 @@
from jwcrypto.jwk import JWK, JWKSet
from pydantic import ValidationError
-from synapse_token_authenticator.claims_validator import Exist, In
+from synapse_token_authenticator.claims_validator import Equal, Exist, In
from synapse_token_authenticator.config.epa import EPaConfig
from tests import get_enc_jwk, get_jwk, get_jwk_set
class TestEPaConfig:
def test_epa_config_defaults(self):
- enc_jwk = get_enc_jwk()
- jwk_set = get_jwk()
config = EPaConfig(
iss="https://example.com",
resource_id="https://example.com",
- enc_jwk=enc_jwk,
- jwk_set=jwk_set,
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk(),
)
assert config.iss == "https://example.com"
assert config.resource_id == "https://example.com"
@@ -32,17 +30,101 @@ def test_epa_config_defaults(self):
assert config.displayname_path is None
assert config.lowercase_localpart is False
+ def test_epa_config_validator(self):
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk(),
+ validator=Exist(),
+ )
+ assert config.validator == Exist()
+
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk(),
+ validator=["equal", "foo"],
+ )
+ assert config.validator == Equal("foo")
+
+ def test_epa_config_jwk_set_as_json_string(self):
+ jwk_str = get_jwk().export()
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_set=jwk_str,
+ )
+ assert isinstance(config.jwk_set, JWK)
+
+ jwk_set_str = get_jwk_set().export()
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_set=jwk_set_str,
+ )
+ assert isinstance(config.jwk_set, JWKSet)
+
+ def test_epa_config_jwk_set_as_dict(self):
+ jwk_dict = get_jwk().export(as_dict=True)
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_set=jwk_dict,
+ )
+ assert isinstance(config.jwk_set, JWK)
+
+ jwk_set_dict = get_jwk_set().export(as_dict=True)
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_set=jwk_set_dict,
+ )
+ assert isinstance(config.jwk_set, JWKSet)
+
+ def test_epa_config_enc_jwk_as_json_string(self):
+ enc_jwk_str = get_enc_jwk().export()
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=enc_jwk_str,
+ jwk_set=get_jwk(),
+ )
+ assert isinstance(config.enc_jwk, JWK)
+
+ def test_epa_config_enc_jwk_set_as_dict(self):
+ enc_jwk_dict = get_enc_jwk().export(as_dict=True)
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=enc_jwk_dict,
+ jwk_set=get_jwk(),
+ )
+ assert isinstance(config.enc_jwk, JWK)
+
def test_epa_config_without_enc_jwk(self):
- jwk_set = get_jwk_set()
with pytest.raises(ValidationError):
EPaConfig(
iss="https://example.com",
resource_id="https://example.com",
- jwk_set=jwk_set,
+ jwk_set=get_jwk_set(),
)
- def test_epa_config_with_enc_jwk_file(self, tmp_path):
- jwk_set = get_jwk_set()
+ def test_epa_config_enc_jwk_file_does_not_exist(self):
+ with pytest.raises(ValidationError):
+ EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk_file="no_such_file.pem",
+ jwk_set=get_jwk(),
+ )
+
+ def test_epa_config_with_enc_jwk_file_opens_and_loads(self, tmp_path):
enc_jwk_path = tmp_path / "enc_jwk.pem"
enc_jwk_path.write_bytes(
JWK.generate(kty="RSA", size=2048).export_to_pem(
@@ -53,13 +135,35 @@ def test_epa_config_with_enc_jwk_file(self, tmp_path):
iss="https://example.com",
resource_id="https://example.com",
enc_jwk_file=str(enc_jwk_path),
- jwk_set=jwk_set,
+ jwk_set=get_jwk_set(),
)
assert isinstance(config.enc_jwk, JWK)
- assert config.enc_jwk_file == str(enc_jwk_path)
+
+ def test_epa_config_jwk_file_does_not_exist(self):
+ with pytest.raises(ValidationError):
+ EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_file="no_such_file.pem",
+ )
+
+ def test_epa_config_with_jwk_file_opens_and_loads(self, tmp_path):
+ jwk_path = tmp_path / "jwk.pem"
+ jwk_path.write_bytes(
+ JWK.generate(kty="RSA", size=2048).export_to_pem(
+ private_key=True, password=None
+ )
+ )
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ enc_jwk=get_enc_jwk(),
+ jwk_file=str(jwk_path),
+ )
+ assert isinstance(config.jwk_set, JWK)
def test_epa_config_more_than_one_enc_jwk_source_should_raise_error(self, tmp_path):
- jwk_set = get_jwk_set()
enc_jwk_path = tmp_path / "enc_jwk.pem"
enc_jwk_path.write_bytes(
JWK.generate(kty="RSA", size=2048).export_to_pem(
@@ -72,30 +176,27 @@ def test_epa_config_more_than_one_enc_jwk_source_should_raise_error(self, tmp_pa
resource_id="https://example.com",
enc_jwk=get_enc_jwk(),
enc_jwk_file=str(enc_jwk_path),
- jwk_set=jwk_set,
+ jwk_set=get_jwk_set(),
)
def test_epa_config_without_jwk_set(self):
- enc_jwk = get_enc_jwk()
with pytest.raises(ValidationError):
EPaConfig(
iss="https://example.com",
resource_id="https://example.com",
- enc_jwk=enc_jwk,
+ enc_jwk=get_enc_jwk(),
)
def test_epa_config_with_more_than_one_jwk_source_should_raise_error(
self, tmp_path
):
- enc_jwk = get_enc_jwk()
- jwk_set = get_jwk_set()
enc_jwk_path = tmp_path / "enc_jwk.pem"
with pytest.raises(ValidationError):
EPaConfig(
iss="https://example.com",
resource_id="https://example.com",
- enc_jwk=enc_jwk,
- jwk_set=jwk_set,
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk_set(),
jwk_file=str(enc_jwk_path),
)
@@ -103,8 +204,8 @@ def test_epa_config_with_more_than_one_jwk_source_should_raise_error(
EPaConfig(
iss="https://example.com",
resource_id="https://example.com",
- enc_jwk=enc_jwk,
- jwk_set=jwk_set,
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk_set(),
jwks_endpoint="https://example.com/.well-known/jwks.json",
)
@@ -112,24 +213,44 @@ def test_epa_config_with_more_than_one_jwk_source_should_raise_error(
EPaConfig(
iss="https://example.com",
resource_id="https://example.com",
- enc_jwk=enc_jwk,
+ enc_jwk=get_enc_jwk(),
jwk_file=str(enc_jwk_path),
jwks_endpoint="https://example.com/.well-known/jwks.json",
)
def test_epa_config_without_iss(self):
- enc_jwk = get_enc_jwk()
- jwk_set = get_jwk_set()
with pytest.raises(ValidationError):
EPaConfig(
resource_id="https://example.com",
- enc_jwk=enc_jwk,
- jwk_set=jwk_set,
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk_set(),
)
- def test_epa_config_does_not_accept_list_expose_metadata_resource(self):
- enc_jwk = get_enc_jwk()
- jwk_set = get_jwk_set()
+ def test_epa_config_with_expose_metadata_resource_none(self):
+ config = EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ expose_metadata_resource=None,
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk_set(),
+ )
+ assert config.expose_metadata_resource is None
+
+ def test_epa_config_fails_with_expose_metadata_resource_name_with_empty_string(
+ self,
+ ):
+ with pytest.raises(
+ ValidationError, match="expose_metadata_resource must have a name field"
+ ):
+ EPaConfig(
+ iss="https://example.com",
+ resource_id="https://example.com",
+ expose_metadata_resource={"name": ""},
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk_set(),
+ )
+
+ def test_epa_config_fails_with_expose_metadata_resource_as_list(self):
with pytest.raises(ValidationError, match="Input should be a valid dictionary"):
EPaConfig(
iss="https://example.com",
@@ -137,8 +258,8 @@ def test_epa_config_does_not_accept_list_expose_metadata_resource(self):
validator=["in", "active", ["equal", True]],
expose_metadata_resource=["something"],
registration_enabled=True,
- enc_jwk=enc_jwk,
- jwk_set=jwk_set,
+ enc_jwk=get_enc_jwk(),
+ jwk_set=get_jwk_set(),
localpart_path="urn:messaging:matrix:localpart",
displayname_path="some_displayname_path",
lowercase_localpart=True,
diff --git a/tests/test_config/test_jwt.py b/tests/test_config/test_jwt.py
index d2f10ea..fd659e9 100644
--- a/tests/test_config/test_jwt.py
+++ b/tests/test_config/test_jwt.py
@@ -20,6 +20,10 @@ def test_jwt_config_wrong_algorithm(self):
with pytest.raises(ValidationError):
JwtConfig(algorithm="invalid")
+ def test_jwt_config_valid_algorithm(self):
+ config = JwtConfig(secret="secret", algorithm="HS256")
+ assert config.algorithm == "HS256"
+
def test_jwt_config_missing_secret_or_keyfile(self):
with pytest.raises(ValidationError):
JwtConfig()
diff --git a/tests/test_config/test_oauth.py b/tests/test_config/test_oauth.py
index 04fe686..1345054 100644
--- a/tests/test_config/test_oauth.py
+++ b/tests/test_config/test_oauth.py
@@ -1,9 +1,10 @@
import pytest
-from jwcrypto.jwk import JWK
+from jwcrypto.jwk import JWK, JWKSet
from pydantic import ValidationError
from synapse_token_authenticator.claims_validator import (
AllOf,
+ Equal,
Exist,
ListAnyOf,
MatchesRegex,
@@ -15,7 +16,7 @@
OAuthConfig,
)
from synapse_token_authenticator.http_auth import BasicAuth, BearerAuth, NoAuth
-from tests import get_jwk
+from tests import get_jwk, get_jwk_set
class TestJwtValidationConfig:
@@ -60,6 +61,31 @@ def test_jwt_validation_config_full(self):
assert config.required_scopes == ["foo", "bar"]
assert isinstance(config.jwk_set, JWK)
+ def test_jwt_validation_config_validator(self):
+ config = JwtValidationConfig(jwk_set=get_jwk(), validator=Exist())
+ assert config.validator == Exist()
+
+ config = JwtValidationConfig(jwk_set=get_jwk(), validator=["equal", "foo"])
+ assert config.validator == Equal("foo")
+
+ def test_jwt_validation_config_jwk_set_as_json_string(self):
+ jwk_str = get_jwk().export()
+ config = JwtValidationConfig(jwk_set=jwk_str)
+ assert isinstance(config.jwk_set, JWK)
+
+ jwk_set_str = get_jwk_set().export()
+ config = JwtValidationConfig(jwk_set=jwk_set_str)
+ assert isinstance(config.jwk_set, JWKSet)
+
+ def test_jwt_validation_config_jwk_set_as_dict(self):
+ jwk_dict = get_jwk().export(as_dict=True)
+ config = JwtValidationConfig(jwk_set=jwk_dict)
+ assert isinstance(config.jwk_set, JWK)
+
+ jwk_set_dict = get_jwk_set().export(as_dict=True)
+ config = JwtValidationConfig(jwk_set=jwk_set_dict)
+ assert isinstance(config.jwk_set, JWKSet)
+
def test_jwt_validation_config_more_than_one_jwk_source_should_raise_error(
self, tmp_path
):
@@ -79,6 +105,17 @@ def test_jwt_validation_config_more_than_one_jwk_source_should_raise_error(
jwks_endpoint="https://example.com/.well-known/jwks.json",
)
+ def test_jwt_validation_config_jwk_file_does_not_exist(self):
+ with pytest.raises(ValidationError):
+ JwtValidationConfig(jwk_file="no_such_file.pem")
+
+ def test_jwt_validation_config_jwk_file_opens_and_loads(self, tmp_path):
+ jwk = JWK.generate(kty="RSA", size=2048)
+ jwk_path = tmp_path / "jwk.pem"
+ jwk_path.write_bytes(jwk.export_to_pem(private_key=True, password=None))
+ config = JwtValidationConfig(jwk_file=str(jwk_path))
+ assert isinstance(config.jwk_set, JWK)
+
def test_jwt_validation_config_required_scopes_accepts_str(self):
config = JwtValidationConfig(jwk_set=get_jwk(), required_scopes="foo bar")
assert config.required_scopes == "foo bar"
diff --git a/tests/test_config/test_oidc.py b/tests/test_config/test_oidc.py
index b1a1031..eb1e955 100644
--- a/tests/test_config/test_oidc.py
+++ b/tests/test_config/test_oidc.py
@@ -50,6 +50,28 @@ def test_oidc_config_allowed_client_ids_accepts_str(self):
)
assert config.allowed_client_ids == ["client-a", "client-b"]
+ def test_oidc_config_allowed_client_ids_accepts_none(self):
+ config = OIDCConfig(
+ issuer="https://example.com",
+ client_id="client_id",
+ client_secret="client_secret",
+ project_id="project_id",
+ organization_id="organization_id",
+ allowed_client_ids=None,
+ )
+ assert config.allowed_client_ids == None
+
+ def test_oidc_config_project_id_and_organization_id_accept_int(self):
+ config = OIDCConfig(
+ issuer="https://example.com",
+ client_id="client_id",
+ client_secret="client_secret",
+ project_id=1234,
+ organization_id=5678,
+ )
+ assert config.project_id == "1234"
+ assert config.organization_id == "5678"
+
def test_oidc_config_is_not_missing_required_fields(self):
with pytest.raises(ValidationError) as e:
OIDCConfig(
diff --git a/tests/test_epa.py b/tests/test_epa.py
index 78ced6a..0ec9a50 100644
--- a/tests/test_epa.py
+++ b/tests/test_epa.py
@@ -13,6 +13,7 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
+import json
from copy import deepcopy
from unittest import mock
@@ -20,6 +21,8 @@
from synapse.types import JsonDict
import tests.unittest as synapsetest
+from synapse_token_authenticator.resources.metadata import MetadataResource
+from synapse_token_authenticator.resources.public_key import PublicKeysResource
from tests import (
ModuleApiTestCase,
get_enc_jwk,
@@ -249,6 +252,45 @@ async def test_localpart_not_lowercase(self):
)
assert result[0] == "@AlIcE:example.test"
+ config_for_epa_metadata = deepcopy(config_for_epa)
+ config_for_epa_metadata["modules"][0]["config"]["epa"][
+ "expose_metadata_resource"
+ ] = {
+ "name": "com.famedly.login.token.epa",
+ "something": "else",
+ }
+
+ @synapsetest.override_config(config_for_epa_metadata)
+ def test_epa_metadata_resource_is_registered(self):
+ path = "/_famedly/login/com.famedly.login.token.epa"
+ resource = self.hs._module_web_resources.get(path)
+ assert isinstance(resource, MetadataResource)
+
+ request = mock.Mock()
+ body = resource.render_GET(request)
+ assert json.loads(body) == {
+ "name": "com.famedly.login.token.epa",
+ "something": "else",
+ }
+ request.setHeader.assert_any_call(b"content-type", b"application/json")
+ request.setHeader.assert_any_call(b"access-control-allow-origin", b"*")
+
+ def test_epa_public_keys_resource_is_registered(self):
+ path = self.hs.mockmod.config.epa.enc_jwks_endpoint
+ resource = self.hs._module_web_resources.get(path)
+ assert path == "/.well-known/jwks.json"
+ assert isinstance(resource, PublicKeysResource)
+
+ request = mock.Mock()
+ body = resource.render_GET(request)
+ published = json.loads(body)
+ expected = jwk.JWKSet()
+ expected.add(get_enc_jwk())
+ assert published == json.loads(expected.export(private_keys=False))
+ assert "d" not in published["keys"][0]
+ request.setHeader.assert_any_call(b"content-type", b"application/json")
+ request.setHeader.assert_any_call(b"access-control-allow-origin", b"*")
+
class CustomFlowTestsWithJwkSet(CustomFlowTests):
"""Same default-config EPA tests, but jwk_set is a JWKS dict (→ JWKSet).
@@ -270,3 +312,4 @@ def default_config(self) -> dict:
test_fetch_jwks = None # type: ignore
test_valid_login_registration_disabled = None # type: ignore
test_localpart_lowercase = None # type: ignore
+ test_epa_metadata_resource_is_registered = None # type: ignore
diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py
index 2d703f8..0ca158c 100644
--- a/tests/test_http_auth.py
+++ b/tests/test_http_auth.py
@@ -11,22 +11,7 @@
)
-class TestHttpAuth:
- def test_parse_auth_invalid_format(self):
- with pytest.raises(
- ValueError, match="Auth parsing failed, expected list or dict"
- ):
- parse_auth("something invalid")
-
- def test_parse_auth_invalid_format_has_single_error_prefix(self):
- with pytest.raises(ValueError) as e:
- parse_auth("something invalid", context="NotifyOnRegistration")
- message = str(e.value)
- assert message.startswith(
- "NotifyOnRegistration: Auth configuration error: Auth parsing failed"
- )
- assert message.count("Auth configuration error:") == 1
-
+class TestHttpAuthModel:
def test_no_auth(self):
no_auth = NoAuth()
assert no_auth.header_map() == {}
@@ -47,32 +32,79 @@ def test_bearer_auth_fail_with_empty_token(self):
with pytest.raises(ValueError, match="String should have at least 1 character"):
BearerAuth(token="")
- def test_parse_dict_auth_none_type_is_no_auth(self):
- assert parse_auth({"type": None}) == NoAuth()
- def test_parse_dict_auth_fail_with_empty_string_type(self):
- with pytest.raises(ValueError, match="Unknown Auth type ''"):
- parse_auth({"type": ""})
+class TestHttpAuthParser:
+ def test_parse_auth_invalid_format(self):
+ with pytest.raises(
+ ValueError, match="Auth parsing failed, expected list or dict"
+ ) as e:
+ parse_auth("something invalid")
+ assert str(e.value).startswith("Auth configuration error: Auth parsing failed")
+
+ # with context
+ with pytest.raises(ValueError) as e:
+ parse_auth("something invalid", context="NotifyOnRegistration")
+ assert str(e.value).startswith(
+ "NotifyOnRegistration: Auth configuration error: Auth parsing failed"
+ )
- def test_parse_dict_auth(self):
+ def test_parse_auth(self):
+ # dict auth
+ assert parse_auth({"type": None}) == NoAuth()
assert parse_auth(
{"type": "basic", "username": "user", "password": "pass"}
) == BasicAuth(username="user", password="pass")
assert parse_auth({"type": "bearer", "token": "token"}) == BearerAuth(
token="token"
)
+ # list auth
+ assert parse_auth([None]) == NoAuth()
+ assert parse_auth(["basic", "user", "pass"]) == BasicAuth(
+ username="user", password="pass"
+ )
+ assert parse_auth(["bearer", "token"]) == BearerAuth(token="token")
+
+ def test_parse_auth_basic_fails_with_missing_type(self):
+ with pytest.raises(
+ ValueError, match="Auth configuration error: pop from empty list"
+ ):
+ parse_auth([])
+ with pytest.raises(ValueError, match="Auth configuration error: 'type'"):
+ parse_auth({})
+
+ def test_parse_auth_fails_with_empty_string_type(self):
+ with pytest.raises(ValueError, match="Unknown Auth type ''"):
+ parse_auth({"type": ""})
+ with pytest.raises(ValueError, match="Unknown Auth type ''"):
+ parse_auth([""])
+
+ def test_parse_auth_fails_with_unknown_auth_type(self):
+ with pytest.raises(ValueError, match="Unknown Auth type 'unknown'"):
+ parse_auth({"type": "unknown", "token": "token"})
+ with pytest.raises(ValueError, match="Unknown Auth type 'unknown'"):
+ parse_auth(["unknown"])
- def test_parse_dict_auth_allows_empty_credentials(self):
+ def test_parse_auth_basic_fails_with_empty_credentials(self):
with pytest.raises(ValueError, match="String should have at least 1 character"):
parse_auth({"type": "basic", "username": "", "password": ""})
with pytest.raises(ValueError, match="String should have at least 1 character"):
- parse_auth({"type": "bearer", "token": ""})
+ parse_auth(["basic", "", ""])
- def test_parse_dict_auth_missing_type(self):
+ def test_parse_auth_basic_fails_with_missing_fields(self):
with pytest.raises(ValueError, match="Auth configuration error: 'type'"):
parse_auth({"username": "user", "password": "pass"})
+ with pytest.raises(ValueError, match="Unknown Auth type 'user'"):
+ parse_auth(["user", "pass"])
+ with pytest.raises(ValueError, match="Field required"):
+ parse_auth({"type": "basic", "password": "pass"})
+ with pytest.raises(ValueError, match="Field required"):
+ parse_auth({"type": "basic"})
+ with pytest.raises(ValueError, match="Field required"):
+ parse_auth(["basic", "user"])
+ with pytest.raises(ValueError, match="Field required"):
+ parse_auth(["basic"])
- def test_parse_dict_auth_basic_extra_fields_are_not_allowed(self):
+ def test_parse_auth_basic_fails_with_extra_fields(self):
with pytest.raises(ValueError, match="Unexpected keyword argument"):
parse_auth(
{
@@ -82,89 +114,31 @@ def test_parse_dict_auth_basic_extra_fields_are_not_allowed(self):
"extra": "field",
}
)
+ with pytest.raises(ValueError, match="Unexpected positional argument"):
+ parse_auth(["basic", "user", "pass", "extra", "field"])
- def test_parse_dict_auth_basic_missing_username(self):
- with pytest.raises(ValueError, match="Field required"):
- parse_auth({"type": "basic", "password": "pass"})
-
- def test_parse_dict_auth_basic_missing_credentials(self):
- with pytest.raises(ValueError, match="Field required"):
- parse_auth({"type": "basic"})
-
- def test_parse_dict_auth_bearer_missing_credentials(self):
- with pytest.raises(ValueError, match="Field required"):
- parse_auth({"type": "bearer"})
-
- def test_parse_dict_auth_unknown_auth_type(self):
- with pytest.raises(ValueError, match="Unknown Auth type 'unknown'"):
- parse_auth({"type": "unknown", "token": "token"})
-
- def test_parse_list_auth_basic_empty_list(self):
- with pytest.raises(
- ValueError, match="Auth configuration error: pop from empty list"
- ):
- parse_auth([])
-
- def test_parse_auth_list(self):
- assert parse_auth([None]) == NoAuth()
- assert parse_auth(["basic", "user", "pass"]) == BasicAuth(
- username="user", password="pass"
- )
- assert parse_auth(["bearer", "token"]) == BearerAuth(token="token")
-
- def test_parse_list_auth_fail_with_empty_credentials(self):
+ def test_parse_auth_bearer_fails_with_empty_credentials(self):
with pytest.raises(ValueError, match="String should have at least 1 character"):
- parse_auth(["basic", "", ""])
+ parse_auth({"type": "bearer", "token": ""})
with pytest.raises(ValueError, match="String should have at least 1 character"):
parse_auth(["bearer", ""])
- def test_parse_list_auth_fail_with_empty_string_type(self):
- with pytest.raises(ValueError, match="Unknown Auth type ''"):
- parse_auth([""])
-
- def test_parse_list_auth_basic_missing_username(self):
- with pytest.raises(ValueError, match="Field required"):
- parse_auth(["basic", "pass"])
-
- def test_parse_list_auth_basic_missing_credentials(self):
+ def test_parse_auth_bearer_fails_with_missing_fields(self):
+ with pytest.raises(ValueError, match="Auth configuration error: 'type'"):
+ parse_auth({"token": "token"})
+ with pytest.raises(ValueError, match="Unknown Auth type 'token'"):
+ parse_auth(["token"])
with pytest.raises(ValueError, match="Field required"):
- parse_auth(["basic"])
-
- def test_parse_list_auth_bearer_missing_credentials(self):
+ parse_auth({"type": "bearer"})
with pytest.raises(ValueError, match="Field required"):
parse_auth(["bearer"])
- def test_parse_list_auth_basic_extra_fields_not_allowed(self):
- with pytest.raises(ValueError, match="Unexpected positional argument"):
- parse_auth(["basic", "user", "pass", "extra", "field"])
-
- def test_parse_list_auth_bearer_extra_fields_not_allowed(self):
+ def test_parse_auth_bearer_fails_with_extra_fields(self):
+ with pytest.raises(ValueError, match="Unexpected keyword argument"):
+ parse_auth({"type": "bearer", "token": "token", "extra": "field"})
with pytest.raises(ValueError, match="Unexpected positional argument"):
parse_auth(["bearer", "token", "extra", "field"])
- def test_parse_list_auth_unknown_auth_type(self):
- with pytest.raises(ValueError, match="Unknown Auth type 'unknown'"):
- parse_auth(["unknown"])
-
- def test_parse_auth_logs_context_for_unknown_type(self, caplog):
- with (
- caplog.at_level(logging.ERROR),
- pytest.raises(ValueError, match="Unknown Auth type 'unknown'"),
- ):
- parse_auth({"type": "unknown"}, context="IntrospectionValidationConfig")
- assert (
- "IntrospectionValidationConfig: Auth configuration error: Unknown Auth type 'unknown'"
- in caplog.text
- )
-
- def test_parse_auth_logs_context_for_missing_credentials(self, caplog):
- with (
- caplog.at_level(logging.ERROR),
- pytest.raises(ValueError, match="Field required"),
- ):
- parse_auth({"type": "basic"}, context="NotifyOnRegistration")
- assert "NotifyOnRegistration: Auth configuration error:" in caplog.text
-
class TestHttpAuthConfigCoercion:
def test_introspection_auth(self):
diff --git a/tests/test_jwt.py b/tests/test_jwt.py
index 33ecc06..4f07598 100644
--- a/tests/test_jwt.py
+++ b/tests/test_jwt.py
@@ -13,8 +13,12 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
+from pathlib import Path
from unittest import mock
+import pytest
+from jwcrypto.jwk import JWK
+
import tests.unittest as synapsetest
from tests import _DEFAULT_TOKEN_SECRET, ModuleApiTestCase, get_jwt_token
@@ -157,3 +161,29 @@ async def test_valid_login_with_admin(self):
self.assertIdentical(
await self.module_api.is_user_admin("@alice:example.test"), True
)
+
+
+class JWTKeyfileTests(ModuleApiTestCase):
+ @pytest.fixture(autouse=True)
+ def _create_jwt_keyfile(self, tmp_path: Path) -> None:
+ self._jwt_key = JWK.generate(kty="RSA", size=2048)
+ keyfile = tmp_path / "jwk.pem"
+ keyfile.write_bytes(
+ self._jwt_key.export_to_pem(private_key=True, password=None)
+ )
+ self._jwt_keyfile = str(keyfile)
+
+ def default_config(self) -> dict:
+ conf = super().default_config()
+ conf["modules"][0]["config"]["jwt"] = {
+ "keyfile": self._jwt_keyfile,
+ "algorithm": "RS256",
+ }
+ return conf
+
+ async def test_valid_login_with_keyfile(self):
+ token = get_jwt_token("alice", algorithm="RS256", key=self._jwt_key)
+ result = await self.hs.mockmod.check_jwt_auth(
+ "alice", "com.famedly.login.token", {"token": token}
+ )
+ assert result[0] == "@alice:example.test"
diff --git a/tests/test_oauth.py b/tests/test_oauth.py
index d4734db..6737dd1 100644
--- a/tests/test_oauth.py
+++ b/tests/test_oauth.py
@@ -13,6 +13,7 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
+import json
from copy import deepcopy
from unittest import mock
@@ -20,6 +21,7 @@
from synapse.types import JsonDict
import tests.unittest as synapsetest
+from synapse_token_authenticator.resources.metadata import MetadataResource
from tests import ModuleApiTestCase, get_jwk, get_jwt_token, mock_for_oauth
default_claims = {
@@ -577,3 +579,26 @@ async def test_login_introspection_threepid(self, add_threepid_mock, *args):
"alice@test.example",
)
assert result[0] == "@alice:example.test"
+
+ config_for_oauth_metadata = deepcopy(config_for_jwt)
+ config_for_oauth_metadata["modules"][0]["config"]["oauth"][
+ "expose_metadata_resource"
+ ] = {
+ "name": "com.famedly.login.token.oauth",
+ "something": "else",
+ }
+
+ @synapsetest.override_config(config_for_oauth_metadata)
+ def test_oauth_metadata_resource_is_registered(self):
+ path = "/_famedly/login/com.famedly.login.token.oauth"
+ resource = self.hs._module_web_resources.get(path)
+ assert isinstance(resource, MetadataResource)
+
+ request = mock.Mock()
+ body = resource.render_GET(request)
+ assert json.loads(body) == {
+ "name": "com.famedly.login.token.oauth",
+ "something": "else",
+ }
+ request.setHeader.assert_any_call(b"content-type", b"application/json")
+ request.setHeader.assert_any_call(b"access-control-allow-origin", b"*")
diff --git a/tests/test_oidc.py b/tests/test_oidc.py
index 0691552..cb4705f 100644
--- a/tests/test_oidc.py
+++ b/tests/test_oidc.py
@@ -13,9 +13,11 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
+import json
from unittest import mock
import tests.unittest as synapsetest
+from synapse_token_authenticator.resources.login_metadata import LoginMetadataResource
from tests import ModuleApiTestCase, get_oidc_login, mock_idp_req
@@ -228,3 +230,21 @@ async def test_allowed_client_ids_space_separated_rejects(self, *args):
"alice", "com.famedly.login.token.oidc", get_oidc_login("alice")
)
assert result is None
+
+ def test_oidc_login_metadata_resource_is_registered(self):
+ path = "/_famedly/login/com.famedly.login.token.oidc"
+ resource = self.hs._module_web_resources.get(path)
+ assert isinstance(resource, LoginMetadataResource)
+
+ request = mock.Mock()
+ body = resource.render_GET(request)
+ assert json.loads(body) == {
+ "issuer": "https://idp.example.test",
+ "issuer-metadata": (
+ "https://idp.example.test/.well-known/openid-configuration"
+ ),
+ "organization-id": "2283783782778",
+ "project-id": "231872387283",
+ }
+ request.setHeader.assert_any_call(b"content-type", b"application/json")
+ request.setHeader.assert_any_call(b"access-control-allow-origin", b"*")
diff --git a/tests/test_validators.py b/tests/test_validators.py
index 5cba4d5..05fd381 100644
--- a/tests/test_validators.py
+++ b/tests/test_validators.py
@@ -3,6 +3,18 @@
from synapse_token_authenticator.claims_validator import parse_validator
+def test_parse_validator_unknown_type():
+ with pytest.raises(ValueError):
+ parse_validator(["unknown", "foo"])
+ with pytest.raises(ValueError):
+ parse_validator({"type": "unknown", "foo": "bar"})
+
+
+def test_parse_validator_invalid_type():
+ with pytest.raises(ValueError):
+ parse_validator("string input")
+
+
def test_validator_exists():
assert parse_validator(["exist"]).validate(None)
@@ -14,6 +26,20 @@ def test_validator_in():
assert not parse_validator(["in", "foo", ["equal", 3]]).validate({"foo": 4})
+def test_validator_in_with_empty_path():
+ with pytest.raises(ValueError):
+ parse_validator(["in", ""])
+ with pytest.raises(ValueError):
+ parse_validator(["in", []])
+ with pytest.raises(ValueError):
+ parse_validator(["in", None])
+
+
+def test_validator_in_with_non_dict_input():
+ assert not parse_validator(["in", "foo", ["equal", 3]]).validate(3)
+ assert not parse_validator(["in", "foo", ["equal", 3]]).validate(["foo", 3])
+
+
def test_validator_not():
assert not parse_validator(["not", ["in", "foo"]]).validate({"foo": 3})
assert parse_validator(["not", ["in", "foo"]]).validate({"loo": 3})
@@ -35,6 +61,16 @@ def test_validator_regex():
assert not parse_validator(["regex", regexp]).validate("bad string")
+def test_validator_regex_non_str():
+ with pytest.raises(ValueError):
+ parse_validator(["regex", 3])
+
+
+def test_validator_regex_invalid_arguments():
+ with pytest.raises(ValueError):
+ parse_validator(["regex", 3, "extra"])
+
+
def test_validator_all_of():
assert parse_validator(["all_of", [["in", "foo"], ["in", "loo"]]]).validate(
{"foo": 3, "loo": 4}
@@ -49,6 +85,16 @@ def test_validator_any_of():
assert parse_validator(["any_of", [["in", "foo"], ["in", "loo"]]]).validate(
{"foo": 3, "loo": 4}
)
+ assert parse_validator(
+ {
+ "type": "any_of",
+ "validators": [
+ {"type": "in", "path": "foo"},
+ {"type": "in", "path": "loo"},
+ ],
+ }
+ ).validate({"foo": 3, "loo": 4})
+
assert parse_validator(["any_of", [["in", "foo"], ["in", "loo"]]]).validate(
{"foo": 3}
)
@@ -58,6 +104,15 @@ def test_validator_any_of():
assert not parse_validator(["any_of", []]).validate({})
+def test_validator_any_of_with_non_dict_input():
+ assert not parse_validator(["any_of", [["in", "foo"], ["in", "loo"]]]).validate(
+ "foo"
+ )
+ assert not parse_validator(["any_of", [["in", "foo"], ["in", "loo"]]]).validate(
+ ["foo", 3]
+ )
+
+
def test_validator_list_all_of():
assert parse_validator(["list_all_of", ["in", "foo"]]).validate(
[{"foo": 3}, {"foo": 4}]
@@ -66,6 +121,16 @@ def test_validator_list_all_of():
assert not parse_validator(["list_all_of", ["in", "foo"]]).validate(
[{"foo": 3}, {"loo": 4}]
)
+ assert parse_validator(
+ {"type": "list_all_of", "validator": {"type": "in", "path": "foo"}}
+ ).validate([{"foo": 3}, {"foo": 4}])
+
+
+def test_validator_list_all_of_with_non_list_input():
+ assert not parse_validator(["list_all_of", ["in", "foo"]]).validate("foo")
+ assert not parse_validator(["list_all_of", ["in", "foo"]]).validate(
+ {"foo": 3, "loo": 4}
+ )
def test_validator_list_any_of():
@@ -76,6 +141,16 @@ def test_validator_list_any_of():
assert parse_validator(["list_any_of", ["in", "foo"]]).validate(
[{"foo": 3}, {"loo": 4}]
)
+ assert parse_validator(
+ {"type": "list_any_of", "validator": {"type": "in", "path": "foo"}}
+ ).validate([{"foo": 3}, {"foo": 4}])
+
+
+def test_validator_list_any_of_with_non_list_input():
+ assert not parse_validator(["list_any_of", ["in", "foo"]]).validate("foo")
+ assert not parse_validator(["list_any_of", ["in", "foo"]]).validate(
+ {"foo": 3, "loo": 4}
+ )
@pytest.fixture