#!/usr/bin/env python3
"""Deserialization detection workflow helper for .NET applications.

This script helps security and engineering teams inventory deserialization
boundaries, classify trust levels, and produce a concise report of risks and
logging controls to review.

It is intentionally non-destructive and does not attempt exploitation.

Usage examples:

  python deserialization_detector.py --input inventory.json
  python deserialization_detector.py --interactive
  python deserialization_detector.py --input inventory.json --output report.md

Input JSON format example:
{
  "services": [
    {
      "name": "orders-api",
      "source": "http",
      "serializer": "System.Text.Json",
      "trusted": false,
      "accepts_type_metadata": false,
      "custom_converters": true,
      "logs_parse_failures": true,
      "notes": "Receives partner payloads"
    }
  ]
}
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Dict, List, Optional


TRUST_LEVELS = {"trusted", "internal", "untrusted"}


@dataclass
class Boundary:
    name: str
    source: str
    serializer: str
    trust_level: str
    accepts_type_metadata: bool = False
    custom_converters: bool = False
    logs_parse_failures: bool = False
    logs_payload_size: bool = False
    logs_rejected_types: bool = False
    notes: str = ""

    def risk_score(self) -> int:
        score = 0
        if self.trust_level == "untrusted":
            score += 3
        elif self.trust_level == "internal":
            score += 1

        if self.accepts_type_metadata:
            score += 3
        if self.custom_converters:
            score += 2
        if not self.logs_parse_failures:
            score += 2
        if not self.logs_payload_size:
            score += 1
        if not self.logs_rejected_types:
            score += 1
        return score

    def risk_label(self) -> str:
        score = self.risk_score()
        if score >= 7:
            return "high"
        if score >= 4:
            return "medium"
        return "low"


def parse_bool(value: Any, field: str) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in {"true", "yes", "1"}:
            return True
        if lowered in {"false", "no", "0"}:
            return False
    raise ValueError(f"Field '{field}' must be a boolean")


def normalize_trust(value: Any) -> str:
    if isinstance(value, bool):
        return "untrusted" if value is False else "trusted"
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in TRUST_LEVELS:
            return lowered
    raise ValueError("Field 'trust_level' must be one of: trusted, internal, untrusted")


def load_inventory(path: Path) -> List[Boundary]:
    data = json.loads(path.read_text(encoding="utf-8"))
    services = data.get("services")
    if not isinstance(services, list):
        raise ValueError("Input JSON must contain a 'services' array")

    boundaries: List[Boundary] = []
    for idx, item in enumerate(services, start=1):
        if not isinstance(item, dict):
            raise ValueError(f"Service entry #{idx} must be an object")

        name = str(item.get("name", "")).strip()
        source = str(item.get("source", "")).strip()
        serializer = str(item.get("serializer", "")).strip()
        if not name or not source or not serializer:
            raise ValueError(f"Service entry #{idx} requires 'name', 'source', and 'serializer'")

        trust_level = item.get("trust_level")
        if trust_level is None and "trusted" in item:
            trust_level = item.get("trusted")
        trust_level = normalize_trust(trust_level)

        boundaries.append(
            Boundary(
                name=name,
                source=source,
                serializer=serializer,
                trust_level=trust_level,
                accepts_type_metadata=parse_bool(item.get("accepts_type_metadata", False), "accepts_type_metadata"),
                custom_converters=parse_bool(item.get("custom_converters", False), "custom_converters"),
                logs_parse_failures=parse_bool(item.get("logs_parse_failures", False), "logs_parse_failures"),
                logs_payload_size=parse_bool(item.get("logs_payload_size", False), "logs_payload_size"),
                logs_rejected_types=parse_bool(item.get("logs_rejected_types", False), "logs_rejected_types"),
                notes=str(item.get("notes", "")).strip(),
            )
        )
    return boundaries


def interactive_inventory() -> List[Boundary]:
    print("Interactive deserialization boundary inventory")
    print("Enter blank name to finish.\n")
    boundaries: List[Boundary] = []
    while True:
        name = input("Boundary name: ").strip()
        if not name:
            break
        source = input("Source (http/queue/file/cache/etc.): ").strip()
        serializer = input("Serializer: ").strip()
        trust_level = normalize_trust(input("Trust level [trusted/internal/untrusted]: ").strip())
        accepts_type_metadata = parse_bool(input("Accepts type metadata? [true/false]: ").strip(), "accepts_type_metadata")
        custom_converters = parse_bool(input("Custom converters? [true/false]: ").strip(), "custom_converters")
        logs_parse_failures = parse_bool(input("Logs parse failures? [true/false]: ").strip(), "logs_parse_failures")
        logs_payload_size = parse_bool(input("Logs payload size? [true/false]: ").strip(), "logs_payload_size")
        logs_rejected_types = parse_bool(input("Logs rejected types? [true/false]: ").strip(), "logs_rejected_types")
        notes = input("Notes (optional): ").strip()
        boundaries.append(
            Boundary(
                name=name,
                source=source,
                serializer=serializer,
                trust_level=trust_level,
                accepts_type_metadata=accepts_type_metadata,
                custom_converters=custom_converters,
                logs_parse_failures=logs_parse_failures,
                logs_payload_size=logs_payload_size,
                logs_rejected_types=logs_rejected_types,
                notes=notes,
            )
        )
        print()
    return boundaries


def render_report(boundaries: List[Boundary]) -> str:
    lines: List[str] = []
    lines.append("# .NET Deserialization Detection Report")
    lines.append("")
    lines.append("This report summarizes deserialization boundaries and highlights detection gaps.")
    lines.append("")

    if not boundaries:
        lines.append("No boundaries were provided.")
        return "\n".join(lines)

    high = sum(1 for b in boundaries if b.risk_label() == "high")
    medium = sum(1 for b in boundaries if b.risk_label() == "medium")
    low = sum(1 for b in boundaries if b.risk_label() == "low")

    lines.append("## Summary")
    lines.append(f"- High risk: {high}")
    lines.append(f"- Medium risk: {medium}")
    lines.append(f"- Low risk: {low}")
    lines.append("")

    for b in sorted(boundaries, key=lambda x: x.risk_score(), reverse=True):
        lines.append(f"### {b.name}")
        lines.append(f"- Source: {b.source}")
        lines.append(f"- Serializer: {b.serializer}")
        lines.append(f"- Trust level: {b.trust_level}")
        lines.append(f"- Risk: {b.risk_label()} (score {b.risk_score()})")
        lines.append(f"- Accepts type metadata: {'yes' if b.accepts_type_metadata else 'no'}")
        lines.append(f"- Custom converters: {'yes' if b.custom_converters else 'no'}")
        lines.append(f"- Logs parse failures: {'yes' if b.logs_parse_failures else 'no'}")
        lines.append(f"- Logs payload size: {'yes' if b.logs_payload_size else 'no'}")
        lines.append(f"- Logs rejected types: {'yes' if b.logs_rejected_types else 'no'}")
        if b.notes:
            lines.append(f"- Notes: {b.notes}")
        lines.append("")

    lines.append("## Recommended review actions")
    lines.append("- Inventory every untrusted deserialization boundary.")
    lines.append("- Reject or strictly allow-list type metadata and polymorphic discriminators.")
    lines.append("- Log parse failures with endpoint, source, payload size, exception category, and trace ID.")
    lines.append("- Avoid logging sensitive payload contents unless policy explicitly allows it.")
    lines.append("- Validate with benign malformed inputs in staging.")
    lines.append("- Confirm alert ownership and incident response before release.")
    lines.append("")
    return "\n".join(lines)


def main(argv: Optional[List[str]] = None) -> int:
    parser = argparse.ArgumentParser(
        description="Inventory .NET deserialization boundaries and generate a detection report."
    )
    parser.add_argument("--input", type=Path, help="Path to a JSON inventory file")
    parser.add_argument("--output", type=Path, help="Write the Markdown report to this file")
    parser.add_argument("--interactive", action="store_true", help="Enter boundary details interactively")
    parser.add_argument("--export-json", type=Path, help="Optional path to export normalized inventory as JSON")

    args = parser.parse_args(argv)

    if not args.input and not args.interactive:
        parser.error("Provide --input or --interactive")

    try:
        boundaries = load_inventory(args.input) if args.input else interactive_inventory()
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    report = render_report(boundaries)

    if args.output:
        args.output.write_text(report, encoding="utf-8")
        print(f"Report written to {args.output}")
    else:
        print(report)

    if args.export_json:
        normalized = {"services": [asdict(b) for b in boundaries]}
        args.export_json.write_text(json.dumps(normalized, indent=2), encoding="utf-8")
        print(f"Normalized inventory written to {args.export_json}")

    return 0


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