cli.py

from __future__ import annotations

import argparse
import json
import sys
from collections import Counter

from .exporters import ExportError, export_redreport
from .parser import SpecError, load_inventory
from .rules import scan_inventory

SEVERITY_ORDER = {"critical": 4, "high": 3, "medium": 2, "low": 1, "informational": 0}


def parser() -> argparse.ArgumentParser:
    root = argparse.ArgumentParser(
        prog="asm", description="Deterministic OpenAPI attack-surface mapper"
    )
    commands = root.add_subparsers(dest="command", required=True)
    validate = commands.add_parser("validate", help="validate and parse an OpenAPI document")
    validate.add_argument("spec")
    inventory = commands.add_parser("inventory", help="list API operations and authentication")
    inventory.add_argument("spec")
    inventory.add_argument("--json", action="store_true")
    scan = commands.add_parser("scan", help="analyze deterministic API security signals")
    scan.add_argument("spec")
    scan.add_argument("--json", action="store_true")
    scan.add_argument("--fail-on", choices=list(SEVERITY_ORDER))
    export = commands.add_parser("export", help="export findings")
    export.add_argument("format", choices=["redreport"])
    export.add_argument("spec")
    export.add_argument("--output", "-o", required=True)
    export.add_argument("--client", default="Portfolio Lab")
    return root


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    try:
        inventory = load_inventory(args.spec)
    except SpecError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    if args.command == "validate":
        print(
            f"VALID: title={inventory.title!r} openapi={inventory.openapi} "
            f"endpoints={len(inventory.endpoints)} servers={len(inventory.servers)}"
        )
        return 0
    if args.command == "inventory":
        values = [
            {
                "method": endpoint.method,
                "path": endpoint.path,
                "operation_id": endpoint.operation_id,
                "anonymous": endpoint.anonymous,
                "auth_schemes": endpoint.auth_schemes,
                "parameters": len(endpoint.parameters),
            }
            for endpoint in inventory.endpoints
        ]
        if args.json:
            print(json.dumps(values, indent=2, ensure_ascii=False))
        else:
            for endpoint in inventory.endpoints:
                auth = "anonymous" if endpoint.anonymous else ",".join(endpoint.auth_schemes)
                print(f"{endpoint.method:7} {endpoint.path:40} auth={auth}")
        return 0

    findings = scan_inventory(inventory)
    if args.command == "scan":
        if args.json:
            print(json.dumps([finding.as_dict() for finding in findings], indent=2))
        else:
            for finding in findings:
                print(
                    f"{finding.severity.upper():6} {finding.rule_id} "
                    f"{finding.asset} — {finding.title}"
                )
            counts = Counter(finding.severity for finding in findings)
            print(
                "SUMMARY "
                + " ".join(f"{severity}={counts[severity]}" for severity in SEVERITY_ORDER)
            )
        if args.fail_on:
            threshold = SEVERITY_ORDER[args.fail_on]
            if any(SEVERITY_ORDER[finding.severity] >= threshold for finding in findings):
                return 3
        return 0
    if args.command == "export":
        try:
            files = export_redreport(inventory, findings, args.output, args.client)
        except ExportError as exc:
            print(f"ERROR: {exc}", file=sys.stderr)
            return 2
        for path in files:
            print(f"EXPORTED: {path.resolve()}")
        return 0
    return 1


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