Merge branch '302-amr' into 'main'

Implement AMR claim

Closes #302

See merge request yaal/canaille!307

Author: azmeuk

CHANGES.rst

@@ -1,3 +1,10 @@
+[0.x.x] - Unreleased
+--------------------
+
+Added
+^^^^^
+- OIDC ``amr`` claim support. :issue:`302`
+
 [0.1.0] - 2025-11-13
 --------------------
 

canaille/app/session.py

@@ -21,6 +21,7 @@
 class UserSession:
     user: User | None = None
     last_login_datetime: datetime.datetime | None = None
+    authentication_methods: list[str] | None = None
 
     @classmethod
     def deserialize(cls, payload):
@@ -38,13 +39,17 @@ def deserialize(cls, payload):
             )
             if payload.get("last_login_datetime")
             else None,
+            authentication_methods=payload.get("authentication_methods"),
         )
 
     def serialize(self):
-        return {
+        payload = {
             "user": self.user.id,
             "last_login_datetime": self.last_login_datetime.isoformat(),
         }
+        if self.authentication_methods is not None:
+            payload["authentication_methods"] = self.authentication_methods
+        return payload
 
 
 def current_user_session():
@@ -67,7 +72,12 @@ def save_user_session() -> None:
 def login_user(user, remember: bool = True) -> None:
     """Open a session for the user."""
     now = datetime.datetime.now(datetime.timezone.utc)
-    obj = UserSession(user=user, last_login_datetime=now)
+    authentication_methods = g.auth.achieved if hasattr(g, "auth") and g.auth else None
+    obj = UserSession(
+        user=user,
+        last_login_datetime=now,
+        authentication_methods=authentication_methods,
+    )
     g.session = obj
     try:
         previous = (

canaille/oidc/provider.py

@@ -44,6 +44,35 @@
 AUTHORIZATION_CODE_LIFETIME = 84400
 JWT_JTI_CACHE_LIFETIME = 3600
 
+AMR_MAPPING = {
+    "password": ["pwd"],
+    "otp": ["otp"],
+    "sms": ["sms", "mca"],
+    "email": ["mca"],
+}
+
+
+def compute_amr_values(authentication_methods):
+    """Convert internal authentication methods to AMR values (RFC 8176).
+
+    Returns a list of AMR values, automatically adding 'mfa' if multiple
+    factors were used.
+    """
+    if not authentication_methods:
+        return None
+
+    amr_values = []
+    for method in authentication_methods:
+        if method in AMR_MAPPING:
+            amr_values.extend(AMR_MAPPING[method])
+
+    amr_values = list(dict.fromkeys(amr_values))
+
+    if len(authentication_methods) > 1:
+        amr_values.append("mfa")
+
+    return amr_values if amr_values else None
+
 
 def get_bearer_token(request):
     """Get the Bearer token from the request headers."""
@@ -86,6 +115,12 @@ def save_authorization_code(code, request):
     nonce = request.payload.data.get("nonce")
     now = datetime.datetime.now(datetime.timezone.utc)
     scope = request.client.get_allowed_scope(request.payload.scope)
+    authentication_methods = (
+        g.session.authentication_methods
+        if hasattr(g, "session") and g.session
+        else None
+    )
+    amr = compute_amr_values(authentication_methods)
     code = models.AuthorizationCode(
         authorization_code_id=gen_salt(48),
         code=code,
@@ -99,6 +134,7 @@ def save_authorization_code(code, request):
         challenge=request.payload.data.get("code_challenge"),
         challenge_method=request.payload.data.get("code_challenge_method"),
         auth_time=g.session.last_login_datetime,
+        amr=amr,
     )
     Backend.instance.save(code)
     return code.code

tests/core/test_auth_password.py

@@ -61,9 +61,13 @@ def test_signin_and_out(testclient, user, caplog):
     res = res.follow(status=200)
 
     with testclient.session_transaction() as session:
-        assert [{"user": user.id, "last_login_datetime": mock.ANY}] == session.get(
-            "sessions"
-        )
+        assert [
+            {
+                "user": user.id,
+                "last_login_datetime": mock.ANY,
+                "authentication_methods": ["password"],
+            }
+        ] == session.get("sessions")
         assert "auth" not in session
 
     res = testclient.get("/login", status=200)
@@ -149,9 +153,13 @@ def test_signin_with_alternate_attribute(testclient, user):
     res = res.follow(status=200)
 
     with testclient.session_transaction() as session:
-        assert [{"user": user.id, "last_login_datetime": mock.ANY}] == session.get(
-            "sessions"
-        )
+        assert [
+            {
+                "user": user.id,
+                "last_login_datetime": mock.ANY,
+                "authentication_methods": ["password"],
+            }
+        ] == session.get("sessions")
 
 
 def test_password_page_without_signin_in_redirects_to_login_page(testclient, user):

tests/oidc/test_amr.py

@@ -0,0 +1,232 @@
+from urllib.parse import parse_qs
+from urllib.parse import urlsplit
+
+from joserfc import jwt
+
+from canaille.app import models
+from canaille.oidc.jose import registry
+from canaille.oidc.provider import compute_amr_values
+from tests.core.test_auth_otp import generate_otp
+
+from . import client_credentials
+
+
+def test_amr_password_only(testclient, user, client, server_jwk, backend):
+    """Test that AMR contains only 'pwd' for password-only authentication."""
+    testclient.app.config["CANAILLE"]["AUTHENTICATION_FACTORS"] = ["password"]
+
+    res = testclient.get("/login")
+    res.form["login"] = "user"
+    res = res.form.submit().follow()
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit()
+
+    res = testclient.get(
+        "/oauth/authorize",
+        params=dict(
+            response_type="code",
+            client_id=client.client_id,
+            scope="openid",
+            nonce="somenonce",
+            redirect_uri="https://client.test/redirect1",
+        ),
+    )
+    res = res.form.submit(name="answer", value="accept", status=302)
+
+    params = parse_qs(urlsplit(res.location).query)
+    code = params["code"][0]
+
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="authorization_code",
+            code=code,
+            scope="openid",
+            redirect_uri=client.redirect_uris[0],
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+    )
+
+    id_token = res.json["id_token"]
+    claims = jwt.decode(id_token, server_jwk, registry=registry)
+    assert claims.claims["amr"] == ["pwd"]
+
+    for consent in backend.query(models.Consent, client=client, subject=user):
+        backend.delete(consent)
+
+
+def test_amr_password_and_otp(testclient, user, client, server_jwk, backend):
+    """Test that AMR contains 'pwd', 'otp', and 'mfa' for password + OTP authentication."""
+    testclient.app.config["CANAILLE"]["AUTHENTICATION_FACTORS"] = ["password", "otp"]
+
+    res = testclient.get("/login")
+    res.form["login"] = "user"
+    res = res.form.submit().follow()
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit().follow()
+
+    totp_period = int(
+        testclient.app.config["CANAILLE"]["TOTP_LIFETIME"].total_seconds()
+    )
+    res.form["otp"] = generate_otp("TOTP", user.secret_token, totp_period=totp_period)
+    res = res.form.submit()
+
+    res = testclient.get(
+        "/oauth/authorize",
+        params=dict(
+            response_type="code",
+            client_id=client.client_id,
+            scope="openid",
+            nonce="somenonce",
+            redirect_uri="https://client.test/redirect1",
+        ),
+    )
+    res = res.form.submit(name="answer", value="accept", status=302)
+
+    params = parse_qs(urlsplit(res.location).query)
+    code = params["code"][0]
+
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="authorization_code",
+            code=code,
+            scope="openid",
+            redirect_uri=client.redirect_uris[0],
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+    )
+
+    id_token = res.json["id_token"]
+    claims = jwt.decode(id_token, server_jwk, registry=registry)
+    assert set(claims.claims["amr"]) == {"pwd", "otp", "mfa"}
+
+    for consent in backend.query(models.Consent, client=client, subject=user):
+        backend.delete(consent)
+
+
+def test_amr_password_and_sms(smtpd, testclient, user, client, server_jwk, backend):
+    """Test that AMR contains 'pwd', 'sms', and 'mfa' for password + SMS authentication."""
+    testclient.app.config["CANAILLE"]["AUTHENTICATION_FACTORS"] = ["password", "sms"]
+
+    res = testclient.get("/login")
+    res.form["login"] = "user"
+    res = res.form.submit().follow()
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit().follow()
+
+    backend.reload(user)
+    res.form["otp"] = user.one_time_password
+    res = res.form.submit()
+
+    res = testclient.get(
+        "/oauth/authorize",
+        params=dict(
+            response_type="code",
+            client_id=client.client_id,
+            scope="openid",
+            nonce="somenonce",
+            redirect_uri="https://client.test/redirect1",
+        ),
+    )
+    res = res.form.submit(name="answer", value="accept", status=302)
+
+    params = parse_qs(urlsplit(res.location).query)
+    code = params["code"][0]
+
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="authorization_code",
+            code=code,
+            scope="openid",
+            redirect_uri=client.redirect_uris[0],
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+    )
+
+    id_token = res.json["id_token"]
+    claims = jwt.decode(id_token, server_jwk, registry=registry)
+    assert set(claims.claims["amr"]) == {"pwd", "sms", "mca", "mfa"}
+
+    for consent in backend.query(models.Consent, client=client, subject=user):
+        backend.delete(consent)
+
+
+def test_amr_password_and_email(smtpd, testclient, user, client, server_jwk, backend):
+    """Test that AMR contains 'pwd', 'mca', and 'mfa' for password + email authentication."""
+    testclient.app.config["CANAILLE"]["AUTHENTICATION_FACTORS"] = [
+        "password",
+        "email",
+    ]
+
+    res = testclient.get("/login")
+    res.form["login"] = "user"
+    res = res.form.submit().follow()
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit().follow()
+
+    backend.reload(user)
+    res.form["otp"] = user.one_time_password
+    res = res.form.submit()
+
+    res = testclient.get(
+        "/oauth/authorize",
+        params=dict(
+            response_type="code",
+            client_id=client.client_id,
+            scope="openid",
+            nonce="somenonce",
+            redirect_uri="https://client.test/redirect1",
+        ),
+    )
+    res = res.form.submit(name="answer", value="accept", status=302)
+
+    params = parse_qs(urlsplit(res.location).query)
+    code = params["code"][0]
+
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="authorization_code",
+            code=code,
+            scope="openid",
+            redirect_uri=client.redirect_uris[0],
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+    )
+
+    id_token = res.json["id_token"]
+    claims = jwt.decode(id_token, server_jwk, registry=registry)
+    assert set(claims.claims["amr"]) == {"pwd", "mca", "mfa"}
+
+    for consent in backend.query(models.Consent, client=client, subject=user):
+        backend.delete(consent)
+
+
+def test_compute_amr_values_none():
+    """Test compute_amr_values with None returns None."""
+    assert compute_amr_values(None) is None
+
+
+def test_compute_amr_values_empty_list():
+    """Test compute_amr_values with empty list returns None."""
+    assert compute_amr_values([]) is None
+
+
+def test_compute_amr_values_unknown_method():
+    """Test compute_amr_values with unknown method returns None."""
+    assert compute_amr_values(["unknown_method"]) is None
+
+
+def test_compute_amr_values_mixed_known_unknown():
+    """Test compute_amr_values with mixed known and unknown methods."""
+    result = compute_amr_values(["password", "unknown_method", "otp"])
+    assert set(result) == {"pwd", "otp", "mfa"}
+
+
+def test_compute_amr_values_deduplication():
+    """Test that AMR values are deduplicated (email and sms both give mca)."""
+    result = compute_amr_values(["email", "sms"])
+    assert result.count("mca") == 1
+    assert set(result) == {"mca", "sms", "mfa"}