truf-server/deploy/fail2ban/edge_e2e_docker_shim.py

#!/usr/bin/python3
"""Constrain production denylist reload commands to the colocated E2E Caddy."""

import os
from pathlib import Path
import re
import stat
import subprocess
import sys


ENV_FILE = Path("/etc/truf-edge/edge.env")
AUDIT_PATH = Path("/var/lib/truf-edge/edge-e2e-reload.audit")
VALIDATION_ERROR_PATH = Path("/var/lib/truf-edge/edge-e2e-validation.error")
COMPOSE_PREFIX = (
    "compose", "--ansi", "never", "--env-file", str(ENV_FILE),
    "--project-directory", "/opt/truf",
    "--file", "/opt/truf/compose.yaml",
    "--file", "/opt/truf/compose.edge.yaml",
)
VALIDATE_COMMAND = COMPOSE_PREFIX + (
    "exec", "-T", "edge", "caddy", "validate",
    "--config", "/etc/caddy/Caddyfile", "--adapter", "caddyfile",
)
RELOAD_COMMAND = COMPOSE_PREFIX + ("kill", "--signal", "SIGUSR1", "edge")
REQUIRED_ENV = {
    "TRUF_EDGE_HOST",
    "TRUF_EDGE_TLS_INCLUDE",
    "TRUF_ADMIN_PREFIX",
    "TRUF_ADMIN_USER",
    "TRUF_ADMIN_PASSWORD_HASH",
    "TRUF_ADMIN_EDGE_MARKER",
}


def classify_command(arguments):
    command = tuple(arguments)
    if command == VALIDATE_COMMAND:
        return "validate"
    if command == RELOAD_COMMAND:
        return "reload"
    raise ValueError("unsupported command")


def audit(operation, result):
    payload = f"{operation}:{result}\n".encode("ascii")
    flags = (
        os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
        | getattr(os, "O_BINARY", 0)
    )
    descriptor = os.open(AUDIT_PATH, flags, 0o600)
    try:
        details = os.fstat(descriptor)
        if not stat.S_ISREG(details.st_mode) or details.st_size + len(payload) > 4096:
            raise ValueError("invalid audit file")
        os.write(descriptor, payload)
    finally:
        os.close(descriptor)


def record_validation_error(content, environment):
    if len(content) > 65536:
        content = b"caddy validation error exceeded evidence bound\n"
    text = content.decode("utf-8", errors="replace")
    for value in environment.values():
        if value:
            text = text.replace(value, "[redacted]")
    text = re.sub(r"\$2[aby]\$(?:0[4-9]|[12][0-9]|3[01])\$[./A-Za-z0-9]{53}", "[redacted]", text)
    text = re.sub(r"(?<![0-9a-f])[0-9a-f]{64}(?![0-9a-f])", "[redacted]", text)
    payload = text.encode("utf-8", errors="replace")[:4096]
    descriptor = os.open(
        VALIDATION_ERROR_PATH,
        os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
        | getattr(os, "O_BINARY", 0),
        0o600,
    )
    try:
        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
            raise ValueError("invalid validation evidence file")
        os.write(descriptor, payload)
    finally:
        os.close(descriptor)


def load_environment(path=ENV_FILE):
    details = path.lstat()
    if not stat.S_ISREG(details.st_mode) or stat.S_ISLNK(details.st_mode) or details.st_size > 8192:
        raise ValueError("invalid environment file")
    values = {}
    for raw_line in path.read_text(encoding="ascii").splitlines():
        if not raw_line or raw_line.startswith("#"):
            continue
        name, separator, value = raw_line.partition("=")
        if not separator or name not in REQUIRED_ENV or name in values or "\x00" in value:
            raise ValueError("invalid environment entry")
        values[name] = value
    if set(values) != REQUIRED_ENV:
        raise ValueError("incomplete environment")
    if (
        values["TRUF_EDGE_HOST"] != "localhost"
        or values["TRUF_EDGE_TLS_INCLUDE"] != "/etc/caddy/tls/static-tls.caddy"
        or not re.fullmatch(r"[0-9a-f]{64}", values["TRUF_ADMIN_PREFIX"])
        or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", values["TRUF_ADMIN_USER"])
        or not re.fullmatch(r"\$2[aby]\$(?:0[4-9]|[12][0-9]|3[01])\$[./A-Za-z0-9]{53}", values["TRUF_ADMIN_PASSWORD_HASH"])
        or not re.fullmatch(r"[0-9a-f]{64}", values["TRUF_ADMIN_EDGE_MARKER"])
    ):
        raise ValueError("unsupported environment")
    return {
        **values,
        "HOME": "/tmp",
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
        "PATH": "/usr/bin:/bin",
    }


def validate():
    try:
        environment = load_environment()
    except Exception:
        audit("environment", 64)
        raise
    try:
        completed = subprocess.run(
            (
                "/usr/bin/caddy", "validate", "--config", "/etc/caddy/Caddyfile",
                "--adapter", "caddyfile",
            ),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            env=environment,
            timeout=30,
            check=False,
        )
    except Exception:
        audit("caddy-exec", 64)
        raise
    if completed.returncode:
        record_validation_error(completed.stderr, environment)
    return completed.returncode


def reload_caddy():
    try:
        command = Path("/proc/1/cmdline").read_bytes()
    except Exception:
        audit("reload-proc", 64)
        raise
    if (
        len(command) > 4096
        or command.rstrip(b"\0").split(b"\0")
        not in (
            [b"caddy", b"run", b"--config", b"/etc/caddy/Caddyfile", b"--adapter", b"caddyfile"],
            [b"/usr/bin/caddy", b"run", b"--config", b"/etc/caddy/Caddyfile", b"--adapter", b"caddyfile"],
        )
    ):
        audit("reload-identity", 64)
        raise ValueError("unexpected pid namespace")
    completed = subprocess.run(
        (
            "/usr/bin/caddy", "reload", "--config", "/etc/caddy/Caddyfile",
            "--adapter", "caddyfile", "--address", "127.0.0.1:2019",
        ),
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=load_environment(),
        timeout=30,
        check=False,
    )
    return completed.returncode


def main(argv=None):
    try:
        operation = classify_command((argv or sys.argv)[1:])
        result = validate() if operation == "validate" else reload_caddy()
    except Exception:
        if "operation" in locals():
            try:
                audit(operation, 64)
            except Exception:
                pass
        return 64
    try:
        audit(operation, result)
    except Exception:
        return 64
    return result


if __name__ == "__main__":
    raise SystemExit(main())