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

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])