net-kingdom/sso-mfa/k8s/lldap/bind-split-rehearsal.py

#!/usr/bin/env python3
# /// script
# dependencies = ["ldap3>=2.9,<3", "PyYAML>=6,<7"]
# ///
"""Synthetic rehearsal of the directory bind split on the pinned LLDAP image (NK-WP-0054-T05).

Agent-runnable: a throwaway LLDAP with generated passwords and a production-shaped directory,
published on 127.0.0.1 only and removed afterwards. No backup, no production data, no Secret.
It answers the questions the live migration depends on and writes a value-free JSON receipt:

- can `ld-admin` be introduced into an existing directory whose super-admin is `admin`, and does
  LLDAP start with `LLDAP_LDAP_USER_DN=ld-admin` without touching `admin`;
- does `admin` keep its rights through `lldap_admin` group membership after that switch;
- does a `lldap_strict_readonly` bind read users and groups but fail to write;
- what is the narrowest group with which the provisioner can create users and set membership;
- can a `svc-*` account sign in interactively to LLDAP (SI-3);
- `--stage authelia`: does the repository's Authelia configuration, bound as `svc-authelia`, admit a
  human and refuse `svc-*` and `ld-admin` (and would the previous filter have admitted them).
"""

from __future__ import annotations

import argparse
import json
import secrets
import subprocess
import sys
import time
import urllib.error
import urllib.request
import uuid
from datetime import UTC, datetime
from pathlib import Path

import ldap3
import yaml

AUTHELIA_IMAGE = "docker.io/authelia/authelia@sha256:bd97cff4fcbf715b5ff1f9ae286afbe6033afce385302520b0368122d43a6f54"
CONFIGMAP = Path(__file__).resolve().parents[1] / "authelia/configmap.yaml"
PREVIOUS_FILTER = "(&(|({username_attribute}={input})({mail_attribute}={input}))(objectClass=inetOrgPerson))"
IMAGE = "docker.io/lldap/lldap@sha256:9e605a66c02514bfcffd1b67cafb1e98d50992216bb2871d7ae44622047dd09d"
BASE_DN = "dc=netkingdom,dc=local"
READ_ONLY = ("svc-authelia", "svc-keycape", "svc-privacyidea")
PRODUCTION_GROUPS = ("net-kingdom-users", "net-kingdom-admins")


def docker(*args, env=None, check=True):
    done = subprocess.run(
        ["docker", *args], check=False, capture_output=True, text=True, env=env
    )
    if check and done.returncode:
        raise RuntimeError(f"docker {args[0]} failed")
    return done


class Directory:
    def __init__(self, tag):
        self.name = f"nk54-bind-split-{tag}"
        self.volume = self.name
        self.jwt = secrets.token_hex(32)
        self.http = self.ldap = None

    def start(self, user_dn, password):
        docker("rm", "--force", self.name, check=False)
        docker(
            "run",
            "--detach",
            "--pull=never",
            "--name",
            self.name,
            "--volume",
            f"{self.volume}:/data",
            "--publish",
            "127.0.0.1::17170",
            "--publish",
            "127.0.0.1::3890",
            "--env",
            f"LLDAP_JWT_SECRET={self.jwt}",
            "--env",
            f"LLDAP_LDAP_BASE_DN={BASE_DN}",
            "--env",
            f"LLDAP_LDAP_USER_DN={user_dn}",
            "--env",
            "LLDAP_LDAP_USER_PASS",
            IMAGE,
            env={"LLDAP_LDAP_USER_PASS": password, "PATH": "/usr/bin:/bin"},
        )
        ports = json.loads(
            docker(
                "inspect", "--format", "{{json .NetworkSettings.Ports}}", self.name
            ).stdout
        )
        self.http = f"http://127.0.0.1:{ports['17170/tcp'][0]['HostPort']}"
        self.ldap = int(ports["3890/tcp"][0]["HostPort"])
        for _ in range(60):
            try:
                with urllib.request.urlopen(self.http + "/health", timeout=2):
                    return
            except (urllib.error.URLError, ConnectionError):
                time.sleep(0.5)
        raise RuntimeError("lldap did not become healthy")

    def stop(self):
        docker("rm", "--force", self.name, check=False)

    def remove(self):
        self.stop()
        docker("volume", "rm", "--force", self.volume, check=False)

    def login(self, user, password):
        request = urllib.request.Request(
            self.http + "/auth/simple/login",
            method="POST",
            data=json.dumps({"username": user, "password": password}).encode(),
            headers={"Content-Type": "application/json"},
        )
        try:
            with urllib.request.urlopen(request, timeout=10) as response:
                return json.load(response)["token"]
        except urllib.error.HTTPError:
            return None

    def graphql(self, token, query, variables=None):
        request = urllib.request.Request(
            self.http + "/api/graphql",
            method="POST",
            data=json.dumps({"query": query, "variables": variables or {}}).encode(),
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {token}",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=10) as response:
                body = json.load(response)
        except urllib.error.HTTPError as error:
            return None, f"http {error.code}"
        return body.get("data"), (body.get("errors") or [{}])[0].get("message")

    def ok(self, token, query, variables=None):
        data, error = self.graphql(token, query, variables)
        if error:
            raise RuntimeError(f"graphql refused: {error}")
        return data

    def group_id(self, token, name):
        groups = self.ok(token, "{ groups { id displayName } }")["groups"]
        return next(g["id"] for g in groups if g["displayName"] == name)

    def create_user(self, token, user, groups=()):
        self.ok(
            token,
            "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }",
            {
                "u": {
                    "id": user,
                    "email": f"{user}@rehearsal.invalid",
                    "displayName": user,
                }
            },
        )
        for group in groups:
            self.ok(
                token,
                "mutation($u: String!, $g: Int!) { addUserToGroup(userId: $u, groupId: $g) { ok } }",
                {"u": user, "g": self.group_id(token, group)},
            )

    def set_password(self, token, user, password):
        docker(
            "exec",
            "--env",
            "LLDAP_USER_PASSWORD",
            self.name,
            "/app/lldap_set_password",
            "--base-url",
            "http://127.0.0.1:17170",
            "--token",
            token,
            "--username",
            user,
            env={"LLDAP_USER_PASSWORD": password, "PATH": "/usr/bin:/bin"},
        )

    def groups_of(self, token, user):
        data = self.ok(
            token,
            "query($u: String!) { user(userId: $u) { groups { displayName } } }",
            {"u": user},
        )
        return sorted(g["displayName"] for g in data["user"]["groups"])

    def bind(self, user, password):
        server = ldap3.Server("127.0.0.1", port=self.ldap, get_info=ldap3.NONE)
        connection = ldap3.Connection(
            server, f"uid={user},ou=people,{BASE_DN}", password
        )
        return connection if connection.bind() else None


def rehearse(tag):
    passwords = {
        name: secrets.token_urlsafe(24)
        for name in ("admin", "ld-admin", "alice", *READ_ONLY, "svc-provisioner")
    }
    directory = Directory(tag)
    result = {}
    try:
        # Production shape: `admin` is the configured super-admin, one human user, the baseline groups.
        directory.start("admin", passwords["admin"])
        admin = directory.login("admin", passwords["admin"])
        for group in PRODUCTION_GROUPS:
            directory.ok(
                admin,
                "mutation($n: String!) { createGroup(name: $n) { id } }",
                {"n": group},
            )
        directory.create_user(admin, "alice", ["net-kingdom-users"])
        directory.set_password(admin, "alice", passwords["alice"])

        # Migration step 1: introduce ld-admin and the service accounts while admin is still configured.
        directory.create_user(admin, "ld-admin", ["lldap_admin"])
        directory.set_password(admin, "ld-admin", passwords["ld-admin"])
        for user in READ_ONLY:
            directory.create_user(admin, user, ["lldap_strict_readonly"])
            directory.set_password(admin, user, passwords[user])
        directory.create_user(admin, "svc-provisioner", ["lldap_password_manager"])
        directory.set_password(admin, "svc-provisioner", passwords["svc-provisioner"])

        # Migration step 2: switch the configured super-admin to ld-admin and restart on the same data.
        directory.stop()
        directory.start("ld-admin", passwords["ld-admin"])
        ld_admin = directory.login("ld-admin", passwords["ld-admin"])
        admin = directory.login("admin", passwords["admin"])
        result["ld_admin_introduced"] = {
            "login": ld_admin is not None,
            "groups": directory.groups_of(ld_admin, "ld-admin") if ld_admin else None,
            "can_create_group": ld_admin is not None
            and directory.graphql(
                ld_admin, 'mutation { createGroup(name: "rehearsal-probe") { id } }'
            )[1]
            is None,
        }
        result["admin_after_switch"] = {
            "login_with_unchanged_password": admin is not None,
            "groups": directory.groups_of(ld_admin, "admin") if ld_admin else None,
            "can_create_user": admin is not None
            and directory.graphql(
                admin,
                "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }",
                {
                    "u": {
                        "id": "probe-by-admin",
                        "email": "probe-by-admin@rehearsal.invalid",
                    }
                },
            )[1]
            is None,
        }

        # Read-only binds: read everything needed by Authelia, KeyCape and the privacyIDEA resolver; write nothing.
        binds = {}
        for user in READ_ONLY:
            connection = directory.bind(user, passwords[user])
            if connection is None:
                binds[user] = {"bind": False}
                continue
            connection.search(
                f"ou=people,{BASE_DN}",
                "(objectClass=person)",
                attributes=["uid", "mail", "memberOf"],
            )
            people = sorted(str(e.uid) for e in connection.entries)
            connection.search(
                f"ou=groups,{BASE_DN}",
                "(objectClass=groupOfUniqueNames)",
                attributes=["cn"],
            )
            groups = len(connection.entries)
            write = connection.modify(
                f"uid=alice,ou=people,{BASE_DN}",
                {"mail": [(ldap3.MODIFY_REPLACE, ["x@rehearsal.invalid"])]},
            )
            connection.unbind()
            binds[user] = {
                "bind": True,
                "people_visible": len(people),
                "sees_alice": "alice" in people,
                "groups_visible": groups,
                "write_refused": not write,
            }
        result["read_only_binds"] = binds

        # Provisioner: what can lldap_password_manager do, and what does lldap_admin add?
        provisioner = directory.login("svc-provisioner", passwords["svc-provisioner"])
        probe = f"probe-{uuid.uuid4().hex[:8]}"
        create_error = directory.graphql(
            provisioner,
            "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }",
            {"u": {"id": probe, "email": f"{probe}@rehearsal.invalid"}},
        )[1]
        users_group = directory.group_id(ld_admin, "net-kingdom-users")
        member_error = directory.graphql(
            provisioner,
            "mutation($u: String!, $g: Int!) { addUserToGroup(userId: $u, groupId: $g) { ok } }",
            {"u": "alice", "g": directory.group_id(ld_admin, "net-kingdom-admins")},
        )[1]
        password_ok = (
            subprocess.run(
                [
                    "docker",
                    "exec",
                    "--env",
                    "LLDAP_USER_PASSWORD",
                    directory.name,
                    "/app/lldap_set_password",
                    "--base-url",
                    "http://127.0.0.1:17170",
                    "--token",
                    provisioner,
                    "--username",
                    "alice",
                ],
                check=False,
                capture_output=True,
                env={
                    "LLDAP_USER_PASSWORD": secrets.token_urlsafe(24),
                    "PATH": "/usr/bin:/bin",
                },
            ).returncode
            == 0
        )
        admin_password_error = (
            subprocess.run(
                [
                    "docker",
                    "exec",
                    "--env",
                    "LLDAP_USER_PASSWORD",
                    directory.name,
                    "/app/lldap_set_password",
                    "--base-url",
                    "http://127.0.0.1:17170",
                    "--token",
                    provisioner,
                    "--username",
                    "ld-admin",
                ],
                check=False,
                capture_output=True,
                env={
                    "LLDAP_USER_PASSWORD": secrets.token_urlsafe(24),
                    "PATH": "/usr/bin:/bin",
                },
            ).returncode
            != 0
        )
        result["provisioner_password_manager"] = {
            "login": provisioner is not None,
            "can_create_user": create_error is None,
            "can_add_group_member": member_error is None,
            "can_set_user_password": password_ok,
            "refused_admin_password": admin_password_error,
            "users_group_id_known": users_group > 0,
        }

        # SI-3: interactive sign-in of a service account to the LLDAP web login.
        result["si3_interactive_login"] = {
            user: directory.login(user, passwords[user]) is not None
            for user in READ_ONLY
        }
    finally:
        directory.remove()
    result["cleanup"] = (
        docker("volume", "inspect", directory.volume, check=False).returncode != 0
    )
    return result


def authelia_config(users_filter, password):
    config = yaml.safe_load(
        yaml.safe_load(CONFIGMAP.read_text())["data"]["configuration.yml"]
    )
    ldap = config["authentication_backend"]["ldap"]
    ldap.pop("url", None)
    ldap.update(address="ldap://lldap:3890", password=password)
    if users_filter is not None:
        ldap["users_filter"] = users_filter
    config.pop("identity_providers", None)
    config["session"]["secret"] = secrets.token_hex(32)
    config["storage"] = {
        "encryption_key": secrets.token_hex(32),
        "local": {"path": "/data/db.sqlite3"},
    }
    config["identity_validation"] = {
        "reset_password": {"jwt_secret": secrets.token_hex(32)}
    }
    config["notifier"] = {"filesystem": {"filename": "/data/notifications.txt"}}
    config["ntp"] = {"disable_startup_check": True}
    config["server"] = {"address": "tcp://0.0.0.0:9091", "disable_healthcheck": True}
    return config, ldap["user"], config["session"]["domain"]


def first_factor(name, config, network, logins, workdir):
    path = workdir / f"{name}.yml"
    path.write_text(yaml.safe_dump(config))
    path.chmod(0o644)
    docker(
        "run",
        "--detach",
        "--pull=never",
        "--name",
        name,
        "--network",
        network,
        "--publish",
        "127.0.0.1::9091",
        "--tmpfs",
        "/data",
        "--volume",
        f"{path}:/config/configuration.yml:ro",
        AUTHELIA_IMAGE,
    )
    try:
        ports = json.loads(
            docker(
                "inspect", "--format", "{{json .NetworkSettings.Ports}}", name
            ).stdout
        )
        base = f"http://127.0.0.1:{ports['9091/tcp'][0]['HostPort']}"
        head = {
            "X-Forwarded-Proto": "https",
            "X-Forwarded-Host": "auth." + config["session"]["domain"],
            "X-Forwarded-For": "127.0.0.1",
            "Content-Type": "application/json",
        }
        for _ in range(120):
            try:
                with urllib.request.urlopen(
                    urllib.request.Request(base + "/api/health", headers=head),
                    timeout=2,
                ):
                    break
            except (urllib.error.URLError, ConnectionError):
                time.sleep(0.5)
        else:
            raise RuntimeError("authelia did not become healthy")
        result = {}
        for label, (user, password) in logins.items():
            body = json.dumps(
                {"username": user, "password": password, "keepMeLoggedIn": False}
            ).encode()
            request = urllib.request.Request(
                base + "/api/firstfactor", data=body, headers=head, method="POST"
            )
            try:
                with urllib.request.urlopen(request, timeout=10) as response:
                    result[label] = response.status
            except urllib.error.HTTPError as error:
                result[label] = error.code
        return result
    finally:
        docker("rm", "--force", name, check=False)


def rehearse_authelia(tag, workdir):
    users = ("alice", "ld-admin", *READ_ONLY)
    passwords = {name: secrets.token_urlsafe(24) for name in ("admin", *users)}
    directory = Directory(tag)
    network = f"nk54-authelia-{tag}"
    result = {}
    try:
        directory.start("admin", passwords["admin"])
        admin = directory.login("admin", passwords["admin"])
        directory.ok(
            admin, 'mutation { createGroup(name: "net-kingdom-users") { id } }'
        )
        directory.create_user(admin, "alice", ["net-kingdom-users"])
        directory.create_user(admin, "ld-admin", ["lldap_admin"])
        for user in READ_ONLY:
            directory.create_user(admin, user, ["lldap_strict_readonly"])
        for user in users:
            directory.set_password(admin, user, passwords[user])
        docker("network", "create", network)
        docker("network", "connect", "--alias", "lldap", network, directory.name)
        logins = {
            "human": ("alice", passwords["alice"]),
            "human_wrong_password": ("alice", "wrong-" + passwords["alice"]),
            "svc-keycape": ("svc-keycape", passwords["svc-keycape"]),
            "svc-privacyidea": ("svc-privacyidea", passwords["svc-privacyidea"]),
            "ld-admin": ("ld-admin", passwords["ld-admin"]),
        }
        config, bind_user, _ = authelia_config(None, passwords["svc-authelia"])
        result["bind_user"] = bind_user
        result["repository_filter"] = first_factor(
            f"nk54-authelia-new-{tag}", config, network, logins, workdir
        )
        config, _, _ = authelia_config(PREVIOUS_FILTER, passwords["svc-authelia"])
        result["previous_filter"] = first_factor(
            f"nk54-authelia-old-{tag}", config, network, logins, workdir
        )
    finally:
        directory.remove()
        docker("network", "rm", network, check=False)
    return result


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--receipt", type=Path, required=True)
    parser.add_argument(
        "--stage", choices=["directory", "authelia"], default="directory"
    )
    args = parser.parse_args(argv)
    if args.receipt.exists():
        print("refused: receipt exists", file=sys.stderr)
        return 2
    tag = uuid.uuid4().hex[:10]
    if args.stage == "authelia":
        import tempfile

        with tempfile.TemporaryDirectory(prefix="nk54-authelia-") as workdir:
            findings = rehearse_authelia(tag, Path(workdir))
    else:
        findings = rehearse(tag)
    receipt = {
        "schema": "netkingdom.lldap-bind-split-rehearsal.v1",
        "task": "NK-WP-0054-T05",
        "observed_at": datetime.now(UTC).isoformat(),
        "image": IMAGE,
        "stage": args.stage,
        "data": "synthetic",
        "production_contacted": False,
        "values_recorded": False,
        "findings": findings,
    }
    args.receipt.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n")
    print(json.dumps(findings, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    sys.exit(main())