user-engine/tests/test_identity_state.py

"""The account site may name a NetKingdom sign-in only after Authelia confirms it."""
import json
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from user_engine.identity_state import (
    AutheliaIdentityState,
    authelia_session_token,
    username_from_state,
    validate_identity_state_url,
    _read_state,
)


class IdentityStateUrlTests(unittest.TestCase):
    def test_accepts_the_cluster_state_endpoint_and_public_https(self):
        cluster = "http://authelia.sso.svc.cluster.local:9091/api/state"
        public = "https://login.coulomb.social/api/state"
        self.assertEqual(cluster, validate_identity_state_url(cluster))
        self.assertEqual(public, validate_identity_state_url(public))

    def test_rejects_anything_that_could_carry_the_session_cookie_elsewhere(self):
        for url in [
            "http://login.coulomb.social/api/state",
            "http://authelia.sso.svc.cluster.local.example/api/state",
            "http://169.254.169.254/api/state",
            "https://user:pass@login.coulomb.social/api/state",
            "https://login.coulomb.social/api/state?next=1",
            "https://login.coulomb.social/api/state#fragment",
            "https://login.coulomb.social/api/userinfo",
            "https://login.coulomb.social/api/state/",
            " https://login.coulomb.social/api/state",
            "http://svc.cluster.local/api/state",
        ]:
            with self.subTest(url=url):
                with self.assertRaises(ValueError):
                    validate_identity_state_url(url)


class IdentityStateParseTests(unittest.TestCase):
    def test_wrapped_and_flat_confirmed_usernames_are_accepted(self):
        wrapped = {
            "status": "OK",
            "data": {"username": "platform-root", "authentication_level": 1},
        }
        flat = {"username": "bernd.worsch-99", "authentication_level": 2}
        self.assertEqual("platform-root", username_from_state(wrapped))
        self.assertEqual("bernd.worsch-99", username_from_state(flat))

    def test_unconfirmed_or_unsafe_answers_are_ignored(self):
        for payload in [
            {"status": "OK", "data": {"username": "", "authentication_level": 0}},
            {"status": "OK", "data": {"username": "platform-root", "authentication_level": 0}},
            {"status": "OK", "data": {"username": "platform-root", "authentication_level": True}},
            {"status": "KO", "data": {"username": "platform-root", "authentication_level": 1}},
            {"username": "<script>", "authentication_level": 1},
            {"username": "platform root", "authentication_level": 1},
            {"data": {"username": "platform-root"}},
            [],
        ]:
            with self.subTest(payload=payload):
                self.assertIsNone(username_from_state(payload))

    def test_cookie_token_rejects_injection_and_other_cookies(self):
        header = "ue_session=keep; authelia_session=opaque.token-1; other=no"
        self.assertEqual("opaque.token-1", authelia_session_token(header))
        self.assertIsNone(authelia_session_token("authelia_session=bad\r\nCookie: x"))
        self.assertIsNone(authelia_session_token("authelia_session=" + ("a" * 4097)))
        self.assertIsNone(authelia_session_token(""))

    def test_lookup_sends_only_the_session_token_and_fails_closed(self):
        seen = []

        def reader(url, token, timeout):
            seen.append((url, token, timeout))
            if token == "broken":
                raise TimeoutError("slow")
            if token == "odd":
                return b'{"status":"OK","data":{"username":"platform-root","authentication_level":1}}'
            return b"not-json"

        state = AutheliaIdentityState(
            "http://authelia.sso.svc.cluster.local:9091/api/state",
            reader=reader,
        )
        header = "ue_session=secret; authelia_session=odd; theme=dark"
        self.assertEqual("platform-root", state.username(header))
        self.assertEqual(
            [("http://authelia.sso.svc.cluster.local:9091/api/state", "odd", 2.0)],
            seen,
        )
        self.assertIsNone(state.username("authelia_session=broken"))
        self.assertIsNone(state.username("authelia_session=plain"))
        self.assertIsNone(state.username("authelia_session=bad\r\nX"))
        self.assertNotIn("secret", json.dumps(seen))


class IdentityStateTransportTests(unittest.TestCase):
    def setUp(self):
        self.seen = []
        parent = self

        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                parent.seen.append((self.path, self.headers.get("Cookie")))
                if self.path == "/api/state":
                    body = json.dumps(
                        {"status": "OK", "data": {"username": "platform-root", "authentication_level": 1}}
                    ).encode()
                    self.send_response(200)
                    self.send_header("Content-Type", "application/json")
                    self.send_header("Content-Length", str(len(body)))
                    self.end_headers()
                    self.wfile.write(body)
                    return
                self.send_response(302)
                self.send_header("Location", "http://127.0.0.1:9/stolen")
                self.end_headers()

            def log_message(self, fmt, *args):
                return

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.port = self.server.server_address[1]

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()

    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)
        self.assertIn(b"platform-root", body)
        self.assertEqual([("/api/state", "authelia_session=opaque-token")], 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])