#!/usr/bin/env python3
"""Create the least-privilege ContextForge API token without exposing it."""

from __future__ import annotations

import argparse
import json
import os
import stat
import tempfile
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlparse

MAX_RESPONSE_BYTES = 1024 * 1024
RUNTIME_PERMISSIONS = (
    "gateways.read",
    "gateways.create",
    "gateways.update",
    "gateways.delete",
    "tools.read",
    "tools.update",
    "tools.execute",
    "prompts.read",
    "servers.read",
    "servers.create",
    "servers.update",
    "servers.use",
)
RUNTIME_READ_CHECKS = (
    "/gateways?limit=1&include_pagination=false",
    "/tools?limit=1&include_pagination=false",
    "/prompts?limit=1&include_pagination=false",
    "/servers?limit=1&include_pagination=false",
)


class BootstrapError(RuntimeError):
    """Sanitized bootstrap failure."""


def _validate_base_url(value: str) -> str:
    parsed = urlparse(value)
    allowed = (
        parsed.scheme == "http"
        and parsed.hostname == "contextforge"
        and parsed.port == 4444
        and not parsed.username
        and not parsed.password
    ) or (
        parsed.scheme == "https"
        and bool(parsed.hostname)
        and not parsed.username
        and not parsed.password
    )
    if not allowed or parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
        raise BootstrapError("ContextForge bootstrap URL is not approved")
    return value.rstrip("/")


def _read_secret(path: Path) -> str:
    try:
        metadata = path.stat()
        payload = path.read_bytes()
    except OSError as exc:
        raise BootstrapError("ContextForge admin password is unavailable") from exc
    content = payload.rstrip(b"\r\n")
    if (
        not stat.S_ISREG(metadata.st_mode)
        or (
            os.name != "nt"
            and (
                stat.S_IMODE(metadata.st_mode) not in {0o400, 0o600}
                or metadata.st_uid != os.getuid()
                or metadata.st_gid != os.getgid()
            )
        )
        or not content
        or len(content) > 8192
        or b"\x00" in content
        or b"\r" in content
        or b"\n" in content
    ):
        raise BootstrapError("ContextForge admin password violates the secret-file policy")
    return content.decode("utf-8")


def _request_json(
    base_url: str,
    method: str,
    path: str,
    *,
    token: str | None = None,
    body: dict[str, Any] | None = None,
) -> Any:
    headers = {"Accept": "application/json"}
    data = None
    if token:
        headers["Authorization"] = f"Bearer {token}"
    if body is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(body, separators=(",", ":")).encode()
    request = urllib.request.Request(  # noqa: S310 - URL is validated before use.
        base_url + path, data=data, headers=headers, method=method
    )
    try:
        with urllib.request.urlopen(request, timeout=10) as response:  # noqa: S310
            payload = response.read(MAX_RESPONSE_BYTES + 1)
            if len(payload) > MAX_RESPONSE_BYTES:
                raise BootstrapError("ContextForge bootstrap response exceeded its limit")
            return json.loads(payload) if payload else None
    except urllib.error.HTTPError as exc:
        raise BootstrapError(
            f"ContextForge {path.split('?', 1)[0]} returned HTTP {exc.code}"
        ) from None
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise BootstrapError("ContextForge bootstrap request failed") from exc


def _write_token(path: Path, token: str) -> None:
    if not token or "\n" in token or "\r" in token or "\x00" in token or len(token) > 8192:
        raise BootstrapError("ContextForge returned an invalid API token")
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    if os.name != "nt" and stat.S_IMODE(path.parent.stat().st_mode) != 0o700:
        raise BootstrapError("ContextForge token directory must have mode 0700")
    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            stream.write(token)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
    except Exception:
        try:
            os.close(descriptor)
        except OSError:
            pass
        try:
            os.unlink(temporary)
        except OSError:
            pass
        raise


def _login(base_url: str, admin_email: str, password: str) -> str:
    login = _request_json(
        base_url,
        "POST",
        "/auth/email/login",
        body={"email": admin_email, "password": password},
    )
    session_token = login.get("access_token") if isinstance(login, dict) else None
    if not isinstance(session_token, str) or not session_token:
        raise BootstrapError("ContextForge login did not return a session token")
    return session_token


def _verify_runtime_token(base_url: str, token: str) -> None:
    for path in RUNTIME_READ_CHECKS:
        _request_json(base_url, "GET", path, token=token)


def _revoke_token(base_url: str, session_token: str, token_id: str) -> None:
    if not token_id:
        raise BootstrapError("ContextForge token identifier was invalid")
    encoded_id = quote(token_id, safe="")
    _request_json(base_url, "DELETE", f"/tokens/{encoded_id}", token=session_token)


def _revoke_named_token(base_url: str, session_token: str, token_name: str) -> None:
    catalog = _request_json(
        base_url,
        "GET",
        "/tokens?include_inactive=false&limit=100&offset=0",
        token=session_token,
    )
    tokens = catalog.get("tokens") if isinstance(catalog, dict) else None
    if not isinstance(tokens, list):
        raise BootstrapError("ContextForge token inventory response was invalid")
    matches = [
        item
        for item in tokens
        if isinstance(item, dict)
        and item.get("name") == token_name
        and item.get("is_active") is True
        and item.get("is_revoked") is not True
    ]
    if len(matches) != 1 or not isinstance(matches[0].get("id"), str) or not matches[0]["id"]:
        raise BootstrapError("ContextForge token rotation requires one active named token")
    _revoke_token(base_url, session_token, matches[0]["id"])


def bootstrap(
    *,
    base_url: str,
    admin_email: str,
    admin_password_file: Path,
    output_token_file: Path,
    source_network: str,
    token_name: str,
    expires_days: int,
    rotate_existing: bool = False,
    replacement_token_name: str | None = None,
) -> dict[str, Any]:
    base_url = _validate_base_url(base_url)
    if rotate_existing and (not replacement_token_name or replacement_token_name == token_name):
        raise BootstrapError("ContextForge rotation requires a distinct replacement token name")
    effective_token_name = replacement_token_name if rotate_existing else token_name
    if not effective_token_name:
        raise BootstrapError("ContextForge token name is required")
    password: str | None = None
    session_token: str | None = None
    if output_token_file.exists():
        existing = _read_secret(output_token_file)
        if rotate_existing:
            password = _read_secret(admin_password_file)
            session_token = _login(base_url, admin_email, password)
            _revoke_named_token(base_url, session_token, token_name)
            try:
                output_token_file.unlink()
            except OSError as exc:
                raise BootstrapError(
                    "ContextForge revoked token file could not be removed"
                ) from exc
        else:
            _verify_runtime_token(base_url, existing)
            return {"status": "pass", "action": "reused", "tokenName": token_name}

    if password is None:
        password = _read_secret(admin_password_file)
    if session_token is None:
        session_token = _login(base_url, admin_email, password)
    created = _request_json(
        base_url,
        "POST",
        "/tokens",
        token=session_token,
        body={
            "name": effective_token_name,
            "description": "ITWorx MCP Hub production gateway reconciliation",
            "expires_in_days": expires_days,
            "scope": {
                "permissions": list(RUNTIME_PERMISSIONS),
                "ip_restrictions": [source_network],
                "time_restrictions": {},
                "usage_limits": {},
            },
            "tags": ["itworx-mcp-hub", "production"],
            "is_active": True,
        },
    )
    api_token = created.get("access_token") if isinstance(created, dict) else None
    token_record = created.get("token") if isinstance(created, dict) else None
    token_id = token_record.get("id") if isinstance(token_record, dict) else None
    if not isinstance(api_token, str) or not api_token:
        try:
            _revoke_named_token(base_url, session_token, effective_token_name)
        except BootstrapError:
            raise BootstrapError(
                "ContextForge token creation returned no API token and cleanup failed"
            ) from None
        raise BootstrapError("ContextForge token creation returned no API token")
    if not isinstance(token_id, str) or not token_id:
        try:
            _revoke_named_token(base_url, session_token, effective_token_name)
        except BootstrapError:
            raise BootstrapError(
                "ContextForge token creation returned no identifier and cleanup failed"
            ) from None
        raise BootstrapError("ContextForge token creation returned no identifier")
    try:
        _verify_runtime_token(base_url, api_token)
        _write_token(output_token_file, api_token)
    except Exception as exc:
        try:
            _revoke_token(base_url, session_token, token_id)
        except BootstrapError:
            raise BootstrapError(
                "ContextForge replacement token validation failed and cleanup failed"
            ) from None
        if isinstance(exc, BootstrapError):
            raise
        raise BootstrapError("ContextForge replacement token could not be stored") from exc
    return {
        "status": "pass",
        "action": "created",
        "tokenName": effective_token_name,
        "permissions": list(RUNTIME_PERMISSIONS),
        "sourceNetwork": source_network,
        "expiresDays": expires_days,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default="http://contextforge:4444")
    parser.add_argument("--admin-email", required=True)
    parser.add_argument("--admin-password-file", type=Path, required=True)
    parser.add_argument("--output-token-file", type=Path, required=True)
    parser.add_argument("--source-network", default="203.0.113.0/24")
    parser.add_argument("--token-name", default="itworx-mcp-hub-runtime-prod")
    parser.add_argument("--expires-days", type=int, default=365)
    parser.add_argument(
        "--rotate-existing",
        action="store_true",
        help=(
            "revoke exactly one invalid named token before creating a distinctly named replacement"
        ),
    )
    parser.add_argument(
        "--replacement-token-name",
        help="unique replacement name required with --rotate-existing",
    )
    args = parser.parse_args()
    if not 1 <= args.expires_days <= 365:
        raise BootstrapError("ContextForge token expiry must be between 1 and 365 days")
    result = bootstrap(
        base_url=args.base_url,
        admin_email=args.admin_email,
        admin_password_file=args.admin_password_file,
        output_token_file=args.output_token_file,
        source_network=args.source_network,
        token_name=args.token_name,
        expires_days=args.expires_days,
        rotate_existing=args.rotate_existing,
        replacement_token_name=args.replacement_token_name,
    )
    print(json.dumps(result, sort_keys=True))
    return 0


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