#!/usr/bin/env python3
"""Redact common secrets and identifiers from CI logs without network access.

This is a defensive convenience tool, not a guarantee that output is anonymous.
It writes to stdout unless --output is supplied and never overwrites input unless
--force is explicitly used.
"""
from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

MAX_BYTES = 20 * 1024 * 1024

RULES: list[tuple[str, re.Pattern[str], str]] = [
    ("private_key", re.compile(r"-----BEGIN(?: [A-Z0-9]+)? PRIVATE KEY-----.*?-----END(?: [A-Z0-9]+)? PRIVATE KEY-----", re.I | re.S), "[REDACTED_PRIVATE_KEY]"),
    ("authorization", re.compile(r"(?im)^(\s*(?:authorization|proxy-authorization)\s*[:=]\s*)(?:bearer|basic|token)\s+\S+"), r"\1[REDACTED_AUTHORIZATION]"),
    ("credential_assignment", re.compile(r"(?i)\b((?:api[_-]?key|access[_-]?token|auth[_-]?token|client[_-]?secret|password|passwd|secret)\s*[:=]\s*)(['\"]?)[^\s,'\";]+\2"), r"\1[REDACTED_SECRET]"),
    ("github_token", re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b"), "[REDACTED_GITHUB_TOKEN]"),
    ("aws_access_key", re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"), "[REDACTED_AWS_ACCESS_KEY]"),
    ("jwt", re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), "[REDACTED_JWT]"),
    ("credential_url", re.compile(r"(?i)(https?://)[^/@\s:]+:[^/@\s]+@"), r"\1[REDACTED_CREDENTIALS]@"),
    ("email", re.compile(r"(?<![A-Za-z0-9._%+-])[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?![A-Za-z0-9.-])"), "[REDACTED_EMAIL]"),
]


def redact(text: str) -> tuple[str, dict[str, int]]:
    counts: dict[str, int] = {}
    for name, pattern, replacement in RULES:
        text, count = pattern.subn(replacement, text)
        if count:
            counts[name] = count
    return text, counts


def read_input(path: str) -> tuple[str, Path | None]:
    if path == "-":
        data = sys.stdin.buffer.read(MAX_BYTES + 1)
        source = None
    else:
        source = Path(path).resolve()
        if not source.is_file():
            raise SystemExit(f"input is not a file: {source}")
        if source.stat().st_size > MAX_BYTES:
            raise SystemExit(f"input exceeds {MAX_BYTES} bytes")
        data = source.read_bytes()
    if len(data) > MAX_BYTES:
        raise SystemExit(f"input exceeds {MAX_BYTES} bytes")
    return data.decode("utf-8", "replace"), source


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", help="CI log path, or - for stdin")
    parser.add_argument("-o", "--output", help="Output path; default is stdout")
    parser.add_argument("--force", action="store_true", help="Allow replacing an existing output file")
    parser.add_argument("--report", action="store_true", help="Print redaction counts to stderr")
    args = parser.parse_args()

    text, source = read_input(args.input)
    cleaned, counts = redact(text)

    if args.output:
        destination = Path(args.output).resolve()
        if source is not None and destination == source and not args.force:
            raise SystemExit("refusing to overwrite input without --force")
        if destination.exists() and not args.force:
            raise SystemExit("output exists; use --force to replace it")
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_text(cleaned, encoding="utf-8")
    else:
        sys.stdout.write(cleaned)

    if args.report:
        total = sum(counts.values())
        details = ", ".join(f"{name}={count}" for name, count in sorted(counts.items())) or "none"
        print(f"redactions={total}; {details}", file=sys.stderr)
        print("Review the result manually: pattern-based redaction cannot guarantee anonymity.", file=sys.stderr)


if __name__ == "__main__":
    main()