From 104405ccbfe37117128719214bc11476c98d3612 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 27 Sep 2026 00:26:50 +0200 Subject: [PATCH] Ask Authelia for sign-in state on the public sign-in host. Authelia rejects the cluster address. Keep the connection inside the cluster and name login.coulomb.social, which matches the session cookie domain. Assistant: grok Assistant-Session: 01a0d25d-d358-7e13-b84a-d007fbb7e34f --- src/user_engine/identity_state.py | 44 +++++++++++++++++++++++-------- src/user_engine/runtime.py | 5 +++- tests/test_identity_state.py | 43 +++++++++++++++++++++++++++--- 3 files changed, 76 insertions(+), 16 deletions(-) diff --git a/src/user_engine/identity_state.py b/src/user_engine/identity_state.py index eff0096..8bb7c2b 100644 --- a/src/user_engine/identity_state.py +++ b/src/user_engine/identity_state.py @@ -18,6 +18,7 @@ from user_engine.oidc import cookie_value _IDENTITY_NAME = re.compile(r"^[A-Za-z0-9._@+-]{1,200}$") _SESSION_TOKEN = re.compile(r"^[A-Za-z0-9._~+/=-]{1,4096}$") +_PUBLIC_HOST = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)+") _CLUSTER_HOST = ".svc.cluster.local" @@ -51,6 +52,13 @@ def validate_identity_state_url(url: str) -> str: return url +def validate_identity_state_host(host: str) -> str: + """Accept the public sign-in host Authelia uses for its session cookie.""" + if not _PUBLIC_HOST.fullmatch(host) or host.endswith(_CLUSTER_HOST): + raise ValueError("identity state host must be the public sign-in hostname") + return host + + def authelia_session_token(cookie_header: str) -> str | None: value = cookie_value(cookie_header, "authelia_session") if value is None or _SESSION_TOKEN.fullmatch(value) is None: @@ -78,15 +86,16 @@ class _RefuseRedirects(HTTPRedirectHandler): raise URLError("identity state endpoint must not redirect") -def _read_state(url: str, token: str, timeout: float) -> bytes: - request = Request( - url, - headers={ - "Accept": "application/json", - "Cookie": f"authelia_session={token}", - }, - method="GET", - ) +def _read_state(url: str, token: str, timeout: float, *, host: str | None = None) -> bytes: + headers = { + "Accept": "application/json", + "Cookie": f"authelia_session={token}", + } + # Authelia rejects the cluster DNS name. The public sign-in host matches + # the session cookie domain, while the connection stays on the cluster URL. + if host: + headers["Host"] = host + request = Request(url, headers=headers, method="GET") opener = build_opener(_RefuseRedirects) with opener.open(request, timeout=timeout) as response: return response.read(8192) @@ -95,10 +104,18 @@ def _read_state(url: str, token: str, timeout: float) -> bytes: class AutheliaIdentityState: """Confirm the username on one Authelia session cookie.""" - def __init__(self, state_url: str, *, timeout: float = 2.0, reader=_read_state) -> None: + def __init__( + self, + state_url: str, + *, + host: str | None = None, + timeout: float = 2.0, + reader=_read_state, + ) -> None: if timeout <= 0: raise ValueError("identity state timeout must be positive") self.state_url = validate_identity_state_url(state_url) + self.host = validate_identity_state_host(host) if host else None self.timeout = timeout self._reader = reader @@ -107,7 +124,12 @@ class AutheliaIdentityState: if token is None: return None try: - raw = self._reader(self.state_url, token, self.timeout) + raw = self._reader( + self.state_url, + token, + self.timeout, + **({"host": self.host} if self.host else {}), + ) payload = json.loads(raw.decode("utf-8")) except (HTTPError, URLError, TimeoutError, OSError, UnicodeError, json.JSONDecodeError, ValueError): return None diff --git a/src/user_engine/runtime.py b/src/user_engine/runtime.py index 33129cb..f1179df 100644 --- a/src/user_engine/runtime.py +++ b/src/user_engine/runtime.py @@ -120,7 +120,10 @@ def create_application() -> PortalApplication: os.environ.get("USER_ENGINE_REGISTRATION_RATE_WINDOW_SECONDS", "60") ), identity_lookup=( - AutheliaIdentityState(os.environ["USER_ENGINE_IDENTITY_STATE_URL"]) + AutheliaIdentityState( + os.environ["USER_ENGINE_IDENTITY_STATE_URL"], + host=os.environ.get("USER_ENGINE_IDENTITY_STATE_HOST") or None, + ) if os.environ.get("USER_ENGINE_IDENTITY_STATE_URL") else None ), diff --git a/tests/test_identity_state.py b/tests/test_identity_state.py index 6f52fbf..cec1671 100644 --- a/tests/test_identity_state.py +++ b/tests/test_identity_state.py @@ -8,6 +8,7 @@ from user_engine.identity_state import ( AutheliaIdentityState, authelia_session_token, username_from_state, + validate_identity_state_host, validate_identity_state_url, _read_state, ) @@ -20,6 +21,19 @@ class IdentityStateUrlTests(unittest.TestCase): self.assertEqual(cluster, validate_identity_state_url(cluster)) self.assertEqual(public, validate_identity_state_url(public)) + def test_public_sign_in_host_is_separate_from_the_cluster_address(self): + self.assertEqual("login.coulomb.social", validate_identity_state_host("login.coulomb.social")) + for host in [ + "authelia.sso.svc.cluster.local", + "login.coulomb.social:443", + "https://login.coulomb.social", + "login", + " login.coulomb.social", + ]: + with self.subTest(host=host): + with self.assertRaises(ValueError): + validate_identity_state_host(host) + def test_rejects_anything_that_could_carry_the_session_cookie_elsewhere(self): for url in [ "http://login.coulomb.social/api/state", @@ -95,6 +109,24 @@ class IdentityStateParseTests(unittest.TestCase): self.assertIsNone(state.username("authelia_session=bad\r\nX")) self.assertNotIn("secret", json.dumps(seen)) + def test_lookup_uses_the_public_sign_in_host(self): + seen = [] + + def reader(url, token, timeout, *, host=None): + seen.append((url, token, host)) + return b'{"status":"OK","data":{"username":"platform-root","authentication_level":1}}' + + state = AutheliaIdentityState( + "http://authelia.sso.svc.cluster.local:9091/api/state", + host="login.coulomb.social", + reader=reader, + ) + self.assertEqual("platform-root", state.username("authelia_session=odd")) + self.assertEqual( + [("http://authelia.sso.svc.cluster.local:9091/api/state", "odd", "login.coulomb.social")], + seen, + ) + class IdentityStateTransportTests(unittest.TestCase): def setUp(self): @@ -103,7 +135,7 @@ class IdentityStateTransportTests(unittest.TestCase): class Handler(BaseHTTPRequestHandler): def do_GET(self): - parent.seen.append((self.path, self.headers.get("Cookie"))) + parent.seen.append((self.path, self.headers.get("Cookie"), self.headers.get("Host"))) if self.path == "/api/state": body = json.dumps( {"status": "OK", "data": {"username": "platform-root", "authentication_level": 1}} @@ -132,10 +164,13 @@ class IdentityStateTransportTests(unittest.TestCase): def test_transport_sends_one_cookie_and_does_not_follow_redirects(self): url = f"http://127.0.0.1:{self.port}/api/state" - body = _read_state(url, "opaque-token", 1) + body = _read_state(url, "opaque-token", 1, host="login.coulomb.social") self.assertIn(b"platform-root", body) - self.assertEqual([("/api/state", "authelia_session=opaque-token")], self.seen) + self.assertEqual( + [("/api/state", "authelia_session=opaque-token", "login.coulomb.social")], + self.seen, + ) with self.assertRaises(Exception): _read_state(f"http://127.0.0.1:{self.port}/redirect", "opaque-token", 1) self.assertEqual("/redirect", self.seen[-1][0]) - self.assertNotIn("/stolen", [path for path, _cookie in self.seen]) + self.assertNotIn("/stolen", [item[0] for item in self.seen])