ITWorx_MCP_Hub-Public/scripts/repair_gitea_creation_scopes.py

"""Explicit Unraid repair for the fixed Hub credential mounts; never prints tokens.

Run on the deployment host with --apply. Keeps root-only rollback copies and
verifies new credentials before changing the mounted files. No repository is created.
"""

import argparse
import json
import os
import re
import subprocess
from datetime import UTC, datetime
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import HTTPRedirectHandler, Request, build_opener

SECRET_ROOT = Path("/srv/example/appdata/itworx-mcp-hub/secrets")
SCOPES = {
    "gitea_mcp_token": "read:issue,read:repository,read:user",
    "gitea_mcp_write_token": "write:issue,write:repository,write:user",
}


class NoCredentialRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def probe(token, path, payload=None):
    request = Request(
        "https://gitea.example.com/api/v1" + path,
        data=None if payload is None else json.dumps(payload).encode(),
        headers={"Authorization": "token " + token, "Content-Type": "application/json"},
    )
    try:
        with build_opener(NoCredentialRedirect()).open(request, timeout=15) as response:
            return response.status, json.load(response)
    except HTTPError as error:
        return error.code, {}


def credential_ready(token, write):
    status, identity = probe(token, "/user")
    return (
        status == 200
        and identity.get("login") == "Jens"
        and probe(token, "/user/repos?limit=1")[0] == 200
        and (not write or probe(token, "/user/repos", {"name": ""})[0] == 422)
    )


def write_mounted(path, content):
    # Preserve the inode: Docker's individual-file mounts use it.
    with path.open("r+b") as stream:
        stream.write(content)
        stream.truncate()
        stream.flush()
        os.fsync(stream.fileno())
    if path.read_bytes() != content:
        raise OSError("Mounted credential verification failed")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--apply", action="store_true")
    args = parser.parse_args()
    if not args.apply:
        print(json.dumps({"apply": False, "scopes": SCOPES}))
        return
    if os.geteuid() != 0 or not SECRET_ROOT.is_dir():
        raise SystemExit("Run only as root on the configured deployment host")
    import fcntl

    lock = (SECRET_ROOT / ".scope-repair.lock").open("a")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        raise SystemExit("Another scope repair is running") from None
    stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    originals = {}
    replacements = {}
    for filename in SCOPES:
        path = SECRET_ROOT / filename
        if path.is_symlink() or not path.is_file():
            raise SystemExit("Expected regular mounted credential file")
        originals[path] = path.read_bytes()
    if all(
        credential_ready(content.decode().strip(), path.name.endswith("write_token"))
        for path, content in originals.items()
    ):
        print(
            json.dumps(
                {
                    "applied": False,
                    "reason": "existing credentials verified",
                    "repositoryCreated": False,
                }
            )
        )
        return
    for filename, scopes in SCOPES.items():
        path = SECRET_ROOT / filename
        result = subprocess.run(  # noqa: S603 -- fixed host command and fixed scopes, no shell
            [
                "/usr/bin/docker",
                "exec",
                "--user",
                "git",
                "Gitea",
                "gitea",
                "admin",
                "user",
                "generate-access-token",
                "--username",
                "Jens",
                "--token-name",
                f"projectbrain-scope-repair-{filename}-{stamp}",
                "--scopes",
                scopes,
                "--raw",
            ],
            capture_output=True,
            timeout=30,
            check=False,
        )
        token = result.stdout.decode().strip()
        if result.returncode or not re.fullmatch(r"[0-9a-f]{40}", token):
            raise SystemExit("Credential generation failed; output withheld")
        if not credential_ready(token, filename.endswith("write_token")):
            raise SystemExit(
                "New credential verification failed; revoke only tokens named "
                f"projectbrain-scope-repair-*-{stamp} after inspection"
            )
        replacements[path] = (token + "\n").encode()
    backup = SECRET_ROOT / ("rollback-scopes-" + stamp)
    backup.mkdir(mode=0o700)
    for path, content in originals.items():
        descriptor = os.open(backup / path.name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(content)
    try:
        for path, content in replacements.items():
            write_mounted(path, content)
        if not all(
            credential_ready(path.read_text().strip(), path.name.endswith("write_token"))
            for path in replacements
        ):
            raise OSError("Installed credential probe failed")
    except Exception:
        failures = []
        for path, content in originals.items():
            try:
                write_mounted(path, content)
            except Exception:
                failures.append(path.name)
        if failures:
            raise SystemExit(
                "Credential install failed; MANUAL RESTORE REQUIRED for "
                + ", ".join(failures)
                + " from "
                + str(backup)
            ) from None
        raise SystemExit(
            "Credential install failed; originals restored. Inspect unused tokens "
            f"named projectbrain-scope-repair-*-{stamp}"
        ) from None
    print(
        json.dumps(
            {
                "applied": True,
                "scopes": SCOPES,
                "rollbackDirectory": str(backup),
                "repositoryCreated": False,
                "restartRequired": False,
            }
        )
    )


if __name__ == "__main__":
    main()