inbuxa-server/tests/e2e/client_override.py

#!/usr/bin/env python3
"""Local end-to-end check that an administrator gets no OAuth client bypass
outside bootstrap and recovery mode (contract C-5).

Run it with `python3 tests/e2e/client_override.py` after
`cargo build -p inbuxa`. Needs Docker. Working state goes under target/e2e.

Administrators hold OAuthClientOverride. Upstream lets it skip the client and
redirect URI checks everywhere, so a link naming a made-up client and an
attacker's redirect URI would hand an administrator's code to the attacker.
This boots the debug binary and checks that:
- in bootstrap mode, the recovery administrator still signs in through an
  unregistered client, as the setup wizard needs;
- after setup, an administrator gets no code for an unregistered client, nor
  for a registered one with a redirect URI it didn't register, while the
  registered client and URI still work end to end;
- a device code an administrator approves for an unregistered client can't be
  exchanged for a token;
- in recovery mode, the bypass is back for the recovery administrator.

Passwords are generated into files under target/e2e and never printed.
Everything is removed afterwards unless KEEP=1.
"""

import base64, hashlib, json, os, secrets, shutil, subprocess, sys, time, urllib.error, urllib.parse, urllib.request

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
DIR = f"{ROOT}/target/e2e"
NAME = "inbuxa-client-override"
PORT = 18195
HTTP = f"http://127.0.0.1:{PORT}"
ADMIN_URL = "http://admin.override.test"
REDIRECT = f"{ADMIN_URL}/oauth/callback"
EVIL = "https://evil.example/cb"
# Another build to check, such as one from before the change.
BINARY = os.environ.get("INBUXA_BINARY", f"{ROOT}/target/debug/inbuxa")

failures = []


def check(cond, what):
    print(("ok   " if cond else "FAIL ") + what)
    if not cond:
        failures.append(what)


def secret_file(name, value=None):
    path = f"{DIR}/secrets/{name}"
    if value is None:
        value = secrets.token_urlsafe(24)
    with open(path, "w") as f:
        f.write(value)
    os.chmod(path, 0o600)
    return value


def docker(*args, check_rc=True):
    return subprocess.run(["docker", *args], capture_output=True, text=True, check=check_rc)


def start(env=None):
    env_file = f"{DIR}/secrets/override-env"
    with open(env_file, "w") as f:
        for key, value in (env or {}).items():
            f.write(f"{key}={value}\n")
    os.chmod(env_file, 0o600)
    docker("run", "-d", "--name", NAME, "--user", f"{os.getuid()}:{os.getgid()}",
           "--entrypoint", "/usr/local/bin/inbuxa",
           "-v", f"{BINARY}:/usr/local/bin/inbuxa:ro",
           "-v", f"{DIR}/etc-override:/etc/inbuxa", "-v", f"{DIR}/data-override:/var/lib/inbuxa",
           "-p", f"127.0.0.1:{PORT}:8080",
           # A debug build's workers need more than the default stack.
           "-e", "RUST_MIN_STACK=16777216",
           # Registers inbuxa-admin, the one client this server knows (C-6).
           "-e", f"INBUXA_ADMIN_URL={ADMIN_URL}",
           "--env-file", env_file,
           "stalwartlabs/stalwart:v0.16.22", "--config", "/etc/inbuxa/config.json")
    for _ in range(120):
        try:
            urllib.request.urlopen(f"{HTTP}/.well-known/jmap", timeout=2)
        except urllib.error.HTTPError:
            return
        except Exception:
            time.sleep(1)
            continue
        return
    sys.exit("server didn't come up: " + docker("logs", "--tail", "40", NAME, check_rc=False).stderr)


def stop():
    docker("rm", "-f", NAME, check_rc=False)


def restart(env=None):
    stop()
    start(env)


def request(path, method="GET", body=None, content_type=None, authorization=None):
    req = urllib.request.Request(f"{HTTP}{path}", data=body, method=method)
    if content_type:
        req.add_header("Content-Type", content_type)
    if authorization:
        req.add_header("Authorization", authorization)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.status, resp.read()
    except urllib.error.HTTPError as err:
        return err.code, err.read()


def jmap(user, password, calls):
    body = json.dumps({"using": ["urn:ietf:params:jmap:core", "urn:inbuxa:jmap:registry"],
                       "methodCalls": calls}).encode()
    auth = "Basic " + base64.b64encode(f"{user}:{password}".encode()).decode()
    status, raw = request("/jmap/", "POST", body, "application/json", auth)
    if status != 200:
        sys.exit(f"JMAP call failed: {status}")
    return json.loads(raw)["methodResponses"]


def pkce():
    verifier = secrets.token_urlsafe(48)
    challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
    return verifier, challenge


def sign_in(user, password, client_id, redirect_uri, challenge):
    """The sign-in page's request: what an authorization link leads to."""
    status, raw = request("/api/auth", "POST", json.dumps({
        "type": "authCode", "accountName": user, "accountSecret": password,
        "clientId": client_id, "redirectUri": redirect_uri,
        "codeChallenge": challenge, "codeChallengeMethod": "S256"}).encode(), "application/json")
    return json.loads(raw) if status == 200 else {"type": status}


def exchange(client_id, code, redirect_uri, verifier):
    status, raw = request("/auth/token", "POST", urllib.parse.urlencode({
        "grant_type": "authorization_code", "client_id": client_id, "code": code,
        "redirect_uri": redirect_uri, "code_verifier": verifier}).encode(),
        "application/x-www-form-urlencoded")
    return status, json.loads(raw or b"{}")


def phished(user, password, client_id, redirect_uri):
    """Whether a link naming this client and redirect URI ends in a token."""
    verifier, challenge = pkce()
    answer = sign_in(user, password, client_id, redirect_uri, challenge)
    if answer.get("type") != "authenticated":
        return False, answer.get("type")
    status, body = exchange(client_id, answer["client_code"], redirect_uri, verifier)
    return status == 200 and "access_token" in body, f"code issued, exchange {status}"


def main():
    stop()
    for sub in ("etc-override", "data-override"):
        shutil.rmtree(f"{DIR}/{sub}", ignore_errors=True)
    for sub in ("etc-override", "data-override", "secrets"):
        os.makedirs(f"{DIR}/{sub}", exist_ok=True)
    os.chmod(f"{DIR}/secrets", 0o700)

    # Bootstrap mode: the recovery administrator keeps the bypass.
    recovery = secret_file("override-recovery")
    start({"INBUXA_RECOVERY_ADMIN": f"admin:{recovery}"})
    got, how = phished("admin", recovery, "setup-wizard", EVIL)
    check(got, f"bootstrap mode: the recovery administrator signs in through an unregistered client ({how})")

    got = jmap("admin", recovery, [["x:Bootstrap/get", {"ids": None}, "0"]])
    singleton = got[0][1]["list"][0]["id"]
    res = jmap("admin", recovery, [["x:Bootstrap/set", {"update": {singleton: {
        "serverHostname": "mail.override.test", "defaultDomain": "override.test",
        "requestTlsCertificate": False}}}, "0"]])
    updated = res[0][1].get("updated", {}).get(singleton)
    check(bool(updated), "bootstrap completed")
    if not updated:
        sys.exit(json.dumps(res))
    admin, admin_pw = updated["username"], secret_file("override-admin", updated["secret"])

    # After setup: no bypass for an administrator.
    restart()
    got, how = phished(admin, admin_pw, "inbuxa-admin", REDIRECT)
    check(got, f"the registered client and redirect URI still sign an administrator in ({how})")
    got, how = phished(admin, admin_pw, "evil-client", EVIL)
    check(not got, f"an unregistered client gets nothing for an administrator ({how})")
    got, how = phished(admin, admin_pw, "inbuxa-admin", EVIL)
    check(not got, f"a registered client with a foreign redirect URI gets nothing ({how})")

    # Device flow: the administrator approves a code a made-up client asked for.
    status, raw = request("/auth/device", "POST", b"client_id=evil-device", "application/x-www-form-urlencoded")
    device = json.loads(raw) if status == 200 else {}
    check("device_code" in device, f"a device code is issued to anyone ({status})")
    if "device_code" in device:
        status, raw = request("/api/auth", "POST", json.dumps({
            "type": "authDevice", "accountName": admin, "accountSecret": admin_pw,
            "code": device["user_code"]}).encode(), "application/json")
        print("     approval:", json.loads(raw).get("type") if status == 200 else status)
        status, raw = request("/auth/token", "POST", urllib.parse.urlencode({
            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
            "client_id": "evil-device", "device_code": device["device_code"]}).encode(),
            "application/x-www-form-urlencoded")
        body = json.loads(raw or b"{}")
        check("access_token" not in body,
              f"but an administrator's approval can't be exchanged for a token ({status}, {body.get('error')})")

    # Recovery mode: the bypass is back, for the recovery administrator.
    restart({"INBUXA_RECOVERY_MODE": "1", "INBUXA_RECOVERY_ADMIN": f"admin:{recovery}"})
    got, how = phished("admin", recovery, "recovery-tool", EVIL)
    check(got, f"recovery mode: the recovery administrator signs in through an unregistered client ({how})")

    if os.environ.get("KEEP") != "1":
        stop()
        for sub in ("etc-override", "data-override"):
            shutil.rmtree(f"{DIR}/{sub}", ignore_errors=True)
        for name in ("override-recovery", "override-admin", "override-env"):
            try:
                os.remove(f"{DIR}/secrets/{name}")
            except FileNotFoundError:
                pass

    print()
    if failures:
        print(f"{len(failures)} failed")
        sys.exit(1)
    print("all passed")


if __name__ == "__main__":
    main()