nexus/tools/cert_checker.py

"""
Certificate Checker Tool - Parse & inspect X.509 certificate files.
Author: Hieu Louis (2026)

Lazy import `cryptography`. Nếu không có, fallback sang `openssl x509`
subprocess. Hỗ trợ PEM và DER.
"""
from __future__ import annotations

import os
import subprocess
from typing import Any, Dict, List, Optional

from .base import Tool, ToolResult, ToolContext, ToolCategory, ToolSafety


class CertCheckerTool(Tool):
    """Parse và inspect X.509 certificate files (.pem, .crt, .der)."""

    category = ToolCategory.SECURITY
    safety = ToolSafety.SAFE

    @property
    def name(self) -> str:
        return "cert_checker"

    @property
    def description(self) -> str:
        return (
            "Parse & inspect X.509 certificate (.pem/.crt/.der). Trả về "
            "subject, issuer, validity, SAN, signature algorithm, serial. "
            "Auto-detect format (PEM/DER) nếu arg `format` không chỉ định."
        )

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Đường dẫn file cert (.pem/.crt/.cer/.der)"},
                "format": {
                    "type": "string",
                    "enum": ["pem", "der", "auto"],
                    "default": "auto",
                    "description": "Format cert. auto = tự detect.",
                },
            },
            "required": ["path"],
        }

    def validate_args(self, args: Dict[str, Any]) -> Optional[str]:
        path = str(args.get("path", "")).strip()
        if not path:
            return "Missing required arg: path"
        if not os.path.exists(path):
            return f"File không tồn tại: {path}"
        if not os.path.isfile(path):
            return f"Path không phải file: {path}"
        fmt = str(args.get("format", "auto"))
        if fmt not in ("pem", "der", "auto"):
            return f"format phải là pem/der/auto, nhận được '{fmt}'"
        return None

    def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult:
        path: str = str(args["path"])
        fmt: str = str(args.get("format", "auto"))

        if context.dry_run:
            return ToolResult(
                success=True,
                output=f"[dry-run] Would inspect cert {path} (format={fmt})",
                metadata={"dry_run": True, "path": path, "format": fmt},
            )

        # Read bytes
        try:
            with open(path, "rb") as f:
                data = f.read()
        except OSError as e:
            return ToolResult(success=False, error=f"Cannot read file: {e}", return_code=1)

        # Auto-detect format
        if fmt == "auto":
            fmt = self._detect_format(data, path)
        if fmt not in ("pem", "der"):
            return ToolResult(
                success=False,
                error=f"Cannot detect format. Hint: pass `format` pem/der.",
                return_code=1,
            )

        # Try cryptography lib first
        try:
            from cryptography import x509  # type: ignore
            from cryptography.hazmat.backends import default_backend  # type: ignore
            cert_obj = self._load_with_cryptography(x509, default_backend, data, fmt)
            info = self._cert_to_dict_cryptography(cert_obj)
            return ToolResult(
                success=True,
                output=self._format_report(info, path, fmt, backend="cryptography"),
                metadata={**info, "path": path, "format": fmt, "backend": "cryptography"},
            )
        except ImportError:
            pass
        except Exception as e:  # noqa: BLE001
            # Nếu cryptography có nhưng parse lỗi → thử openssl fallback
            last_err = str(e)
        else:
            last_err = None

        # Fallback: openssl CLI
        info, err = self._parse_with_openssl(path, fmt, context.timeout or 15)
        if info is not None:
            return ToolResult(
                success=True,
                output=self._format_report(info, path, fmt, backend="openssl CLI"),
                metadata={**info, "path": path, "format": fmt, "backend": "openssl"},
            )
        return ToolResult(
            success=False,
            error=(
                f"Không parse được cert. cryptography không khả dụng hoặc "
                f"lỗi parse ({last_err}); openssl CLI fallback cũng thất bại: {err}"
            ),
            return_code=1,
        )

    # ---------- Format detection ----------
    @staticmethod
    def _detect_format(data: bytes, path: str) -> str:
        """PEM nếu có '-----BEGIN CERTIFICATE-----', ngược lại DER."""
        head = data[:64].lstrip()
        if head.startswith(b"-----BEGIN"):
            return "pem"
        ext = os.path.splitext(path)[1].lower()
        if ext in (".der", ".cer"):
            return "der"
        if ext in (".pem", ".crt"):
            return "pem"
        # Heuristic: DER thường bắt đầu bằng 0x30 (ASN.1 SEQUENCE)
        if data[:1] == b"\x30":
            return "der"
        return "pem"  # default

    # ---------- cryptography lib ----------
    @staticmethod
    def _load_with_cryptography(
        x509_module: Any, backend: Any, data: bytes, fmt: str
    ) -> Any:
        if fmt == "pem":
            return x509_module.load_pem_x509_certificate(data, backend)
        return x509_module.load_der_x509_certificate(data, backend)

    def _cert_to_dict_cryptography(self, cert: Any) -> Dict[str, Any]:
        from cryptography import x509 as _x509  # type: ignore
        from cryptography.x509.oid import ExtensionOID  # type: ignore

        not_before = getattr(cert, "not_valid_before_utc", None) or cert.not_valid_before
        not_after = getattr(cert, "not_valid_after_utc", None) or cert.not_valid_after

        san_list: List[str] = []
        try:
            ext = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
            san = ext.value
            san_list = san.get_values_for_type(_x509.DNSName)
            try:
                for ip in san.get_values_for_type(_x509.IPAddress):
                    san_list.append(str(ip))
            except Exception:
                pass
        except _x509.ExtensionNotFound:
            pass

        sig_algo = None
        try:
            sig_algo = cert.signature_algorithm_oid._name
        except Exception:
            sig_algo = str(cert.signature_hash_algorithm)

        return {
            "subject": cert.subject.rfc4514_string(),
            "issuer": cert.issuer.rfc4514_string(),
            "not_before": not_before.isoformat() if not_before else None,
            "not_after": not_after.isoformat() if not_after else None,
            "san": san_list,
            "signature_algorithm": sig_algo,
            "serial_number": format(cert.serial_number, "x"),
            "version": str(cert.version),
            "is_ca": self._is_ca(cert, _x509, ExtensionOID),
        }

    @staticmethod
    def _is_ca(cert: Any, x509_module: Any, ExtensionOID: Any) -> bool:
        try:
            ext = cert.extensions.get_extension_for_oid(ExtensionOID.BASIC_CONSTRAINTS)
            return bool(ext.value.ca)
        except Exception:
            return False

    # ---------- openssl CLI fallback ----------
    def _parse_with_openssl(
        self, path: str, fmt: str, timeout: int
    ) -> tuple[Optional[Dict[str, Any]], Optional[str]]:
        cmd = ["openssl", "x509", "-in", path, "-noout", "-text"]
        if fmt == "der":
            cmd.insert(2, "-inform")
            cmd.insert(3, "DER")
        try:
            res = subprocess.run(
                cmd, capture_output=True, text=True, timeout=timeout, check=False
            )
        except FileNotFoundError:
            return None, "openssl CLI không có trên PATH"
        except subprocess.TimeoutExpired:
            return None, "openssl timeout"
        except Exception as e:  # noqa: BLE001
            return None, str(e)

        if res.returncode != 0:
            return None, res.stderr.strip() or "openssl parse error"

        text = res.stdout
        info: Dict[str, Any] = {
            "subject": self._grep_field(text, "Subject:"),
            "issuer": self._grep_field(text, "Issuer:"),
            "not_before": self._grep_field(text, "Not Before:"),
            "not_after": self._grep_field(text, "Not After:"),
            "signature_algorithm": self._grep_field(text, "Signature Algorithm:") or None,
            "serial_number": self._grep_field(text, "Serial Number:"),
            "san": [],
            "version": self._grep_field(text, "Version:"),
            "is_ca": False,
        }
        # Try SAN extraction
        san_line = self._grep_field(text, "DNS:")
        if san_line:
            info["san"] = [s.strip() for s in san_line.split(",") if s.strip().startswith("DNS:")]
            info["san"] = [s.replace("DNS:", "").strip() for s in info["san"]]
        info["is_ca"] = "CA:TRUE" in text
        return info, None

    @staticmethod
    def _grep_field(text: str, key: str) -> Optional[str]:
        for line in text.splitlines():
            if key in line:
                return line.split(":", 1)[-1].strip() or None
        return None

    # ---------- Report ----------
    @staticmethod
    def _format_report(info: Dict[str, Any], path: str, fmt: str, backend: str) -> str:
        san = info.get("san") or []
        lines = [
            f"Certificate inspection ({fmt.upper()}, backend={backend}):",
            f"  Path     : {path}",
            f"  Subject  : {info.get('subject')}",
            f"  Issuer   : {info.get('issuer')}",
            f"  Not before: {info.get('not_before')}",
            f"  Not after : {info.get('not_after')}",
            f"  Serial   : {info.get('serial_number')}",
            f"  Version  : {info.get('version')}",
            f"  Sig algo : {info.get('signature_algorithm')}",
            f"  Is CA    : {info.get('is_ca')}",
            f"  SAN ({len(san)}):",
        ]
        for s in san:
            lines.append(f"    - {s}")
        return "\n".join(lines)