#!/usr/bin/env python3 from __future__ import annotations import json import re from collections import Counter, defaultdict from pathlib import Path from typing import Any import yaml ROOT = Path("/Users/lc/coding/pingcap-kb-source") APP_DELIVERY = Path("/Users/lc/coding/app-delivery") MAPPING_JSON = ROOT / "knowledge/deployment/app-delivery-object-mapping.json" OUTPUT_JSON = ROOT / "knowledge/deployment/public-entrypoint-catalog.json" OUTPUT_MD = ROOT / "knowledge/deployment/public-entrypoint-scan.md" SKIP_PATH_PARTS = { "charts", "crds", "ci", "upstream", } SKIP_FILE_PATTERNS = ( "argocd-install-", "_tmpl.yaml", "values.yaml", ) MAX_FILE_SIZE = 500_000 KIND_PATTERN = re.compile(r"^kind:\s*(Ingress|Service|Gateway|HTTPRoute|GRPCRoute)\s*$", re.MULTILINE) PUBLIC_HOST_HINTS = ( "tidbcloud.com", "tidbapi.com", ) PRIVATE_HOST_HINTS = ( "privatelink", ".internal", ".cluster.local", ) EXAMPLE_HOST_HINTS = ( "chart-example.local", "example.com", "example.local", ) OVERLAY_ROOTS = { "dev", "staging", } def load_mapping() -> tuple[dict[str, dict[str, Any]], dict[str, list[str]]]: with MAPPING_JSON.open() as f: mapping_data = json.load(f) object_mapping: dict[str, dict[str, Any]] = {} interface_files: dict[str, list[str]] = {} for item in mapping_data["objects"]: deployment_object = item["deployment_object"] object_mapping[deployment_object] = item source_owner = item.get("source_repo") or item.get("source_family") if not source_owner: continue repo_name = source_owner.split("/")[-1] repo_dir = ROOT / "knowledge/repos" / repo_name / "interfaces" if repo_dir.exists(): interface_files[source_owner] = sorted( str(path.relative_to(ROOT)) for path in repo_dir.glob("*.json") ) else: interface_files[source_owner] = [] return object_mapping, interface_files def iter_yaml_files() -> list[Path]: files: list[Path] = [] for path in APP_DELIVERY.rglob("*.yaml"): rel = path.relative_to(APP_DELIVERY) rel_str = str(rel) if any(part in SKIP_PATH_PARTS for part in rel.parts): continue if any(pattern in rel_str for pattern in SKIP_FILE_PATTERNS): continue try: if path.stat().st_size > MAX_FILE_SIZE: continue except OSError: continue files.append(path) return files def is_templated(value: str | None) -> bool: if not value: return False return "{{" in value or "}}" in value def is_example_host(host: str | None) -> bool: if not host: return False return any(hint in host for hint in EXAMPLE_HOST_HINTS) def classify_host(host: str | None) -> str: if not host: return "unknown" if is_templated(host) or is_example_host(host): return "unknown" if any(hint in host for hint in PRIVATE_HOST_HINTS): return "private_dns" if any(hint in host for hint in PUBLIC_HOST_HINTS): return "public_dns" return "unknown_dns" def infer_environment_hint(rel_parts: tuple[str, ...]) -> str | None: labels = ("envs", "env", "variants", "alicloud-envs") for idx, part in enumerate(rel_parts): if part in labels and idx + 1 < len(rel_parts): return rel_parts[idx + 1] return None def annotation_lookup(metadata: dict[str, Any], key: str) -> str | None: annotations = metadata.get("annotations") or {} value = annotations.get(key) return None if value is None else str(value) def classify_service_reachability(metadata: dict[str, Any], spec: dict[str, Any], service_type: str) -> str: if service_type == "NodePort": return "unknown" scheme = annotation_lookup(metadata, "service.beta.kubernetes.io/aws-load-balancer-scheme") internal = annotation_lookup(metadata, "service.beta.kubernetes.io/aws-load-balancer-internal") if scheme == "internal" or internal in {"true", "0.0.0.0/0"}: return "internal_lb" if scheme == "internet-facing": return "internet_facing_lb" return "unknown_lb" def detect_protocol(ports: list[dict[str, Any]]) -> str: names = [str(p.get("name", "")).lower() for p in ports] if any("grpc" in name for name in names): return "grpc" if any("http" in name or "web" in name for name in names): return "http" return "tcp" def resolve_source_owner(mapping: dict[str, Any] | None) -> str | None: if not mapping: return None return mapping.get("source_repo") or mapping.get("source_family") def infer_deployment_object(rel_path: Path) -> str: if rel_path.parts[0] in OVERLAY_ROOTS and len(rel_path.parts) > 1: return rel_path.parts[1] return rel_path.parts[0] def build_base_record( rel_path: Path, metadata: dict[str, Any], exposure_surface: str, deployment_object: str, mapping: dict[str, Any] | None, repo_interfaces: dict[str, list[str]], ) -> dict[str, Any]: source_repo = resolve_source_owner(mapping) interface_artifacts = repo_interfaces.get(source_repo or "", []) return { "deployment_object": deployment_object, "component_path": str(rel_path.parent), "source_repo": source_repo, "mapping_type": mapping.get("mapping_type") if mapping else None, "mapping_confidence": mapping.get("confidence") if mapping else None, "exposure_surface": exposure_surface, "manifest_name": metadata.get("name"), "namespace": metadata.get("namespace"), "environment_hint": infer_environment_hint(rel_path.parts), "source_file": str(rel_path), "interface_artifacts": interface_artifacts, "notes": [], } def normalize_path_value(match: dict[str, Any]) -> str | None: path = match.get("path") if isinstance(path, dict): return path.get("value") if isinstance(path, str): return path return None def main() -> None: object_mapping, repo_interfaces = load_mapping() records: list[dict[str, Any]] = [] yaml_files = iter_yaml_files() for path in yaml_files: rel_path = path.relative_to(APP_DELIVERY) deployment_object = infer_deployment_object(rel_path) mapping = object_mapping.get(deployment_object) try: content = path.read_text() except UnicodeDecodeError: continue if not KIND_PATTERN.search(content): continue if "{{" in content and "kind:" not in content: continue try: docs = list(yaml.safe_load_all(content)) except Exception: continue for doc in docs: if not isinstance(doc, dict): continue kind = doc.get("kind") metadata = doc.get("metadata") or {} spec = doc.get("spec") or {} if not kind or not isinstance(metadata, dict) or not isinstance(spec, dict): continue if kind == "Ingress": ingress_class = spec.get("ingressClassName") or annotation_lookup(metadata, "kubernetes.io/ingress.class") tls = spec.get("tls") or [] for rule in spec.get("rules") or []: host = rule.get("host") if is_example_host(host): continue http = rule.get("http") or {} paths = http.get("paths") or [{}] for index, path_rule in enumerate(paths): backend = ((path_rule.get("backend") or {}).get("service") or {}) service_port = backend.get("port") or {} record = build_base_record(rel_path, metadata, "ingress", deployment_object, mapping, repo_interfaces) path_value = path_rule.get("path") record.update( { "entrypoint_id": f"entry:{deployment_object}:ingress:{metadata.get('name')}:{index}", "reachability": classify_host(host), "host": host, "path": path_value, "port": service_port.get("number") or service_port.get("name"), "protocol": "http", "backend_service": backend.get("name"), "evidence_kind": "Ingress", "ingress_class": ingress_class, "tls_enabled": bool(tls), } ) if is_templated(host): record["notes"].append("templated_host") records.append(record) elif kind == "HTTPRoute": hostnames = spec.get("hostnames") or [None] parent_refs = spec.get("parentRefs") or [] for rule_index, rule in enumerate(spec.get("rules") or []): matches = rule.get("matches") or [{}] backend_refs = rule.get("backendRefs") or [] for match in matches: match_path = normalize_path_value(match) for backend in backend_refs: for host in hostnames: record = build_base_record(rel_path, metadata, "httproute", deployment_object, mapping, repo_interfaces) reachability = classify_host(host) if host else "gateway_routed_unknown_host" record.update( { "entrypoint_id": f"entry:{deployment_object}:httproute:{metadata.get('name')}:{rule_index}:{backend.get('name')}", "reachability": reachability, "host": host, "path": match_path, "port": backend.get("port"), "protocol": "http", "backend_service": backend.get("name"), "evidence_kind": "HTTPRoute", "gateway_parents": parent_refs, } ) records.append(record) elif kind == "GRPCRoute": hostnames = spec.get("hostnames") or [None] parent_refs = spec.get("parentRefs") or [] for rule_index, rule in enumerate(spec.get("rules") or []): backend_refs = rule.get("backendRefs") or [] for backend in backend_refs: for host in hostnames: record = build_base_record(rel_path, metadata, "grpcroute", deployment_object, mapping, repo_interfaces) reachability = classify_host(host) if host else "gateway_routed_unknown_host" record.update( { "entrypoint_id": f"entry:{deployment_object}:grpcroute:{metadata.get('name')}:{rule_index}:{backend.get('name')}", "reachability": reachability, "host": host, "path": None, "port": backend.get("port"), "protocol": "grpc", "backend_service": backend.get("name"), "evidence_kind": "GRPCRoute", "gateway_parents": parent_refs, } ) records.append(record) elif kind == "Gateway": listeners = spec.get("listeners") or [] for index, listener in enumerate(listeners): host = listener.get("hostname") record = build_base_record(rel_path, metadata, "gateway", deployment_object, mapping, repo_interfaces) reachability = classify_host(host) if host else "gateway_routed_unknown_host" record.update( { "entrypoint_id": f"entry:{deployment_object}:gateway:{metadata.get('name')}:{index}", "reachability": reachability, "host": host, "path": None, "port": listener.get("port"), "protocol": listener.get("protocol"), "backend_service": None, "evidence_kind": "Gateway", "gateway_listener_name": listener.get("name"), } ) records.append(record) elif kind == "Service": service_type = spec.get("type") if service_type not in {"LoadBalancer", "NodePort"}: continue ports = spec.get("ports") or [] record = build_base_record( rel_path, metadata, "loadbalancer_service" if service_type == "LoadBalancer" else "nodeport_service", deployment_object, mapping, repo_interfaces, ) annotations = metadata.get("annotations") or {} record.update( { "entrypoint_id": f"entry:{deployment_object}:service:{metadata.get('name')}", "reachability": classify_service_reachability(metadata, spec, service_type), "host": None, "path": None, "port": [p.get("port") for p in ports], "protocol": detect_protocol(ports), "backend_service": metadata.get("name"), "evidence_kind": "Service", "service_type": service_type, "service_annotations": annotations, } ) records.append(record) exposure_surface_counts = Counter(record["exposure_surface"] for record in records) reachability_counts = Counter(record["reachability"] for record in records) deployment_object_counts = Counter(record["deployment_object"] for record in records) internet_reachable_records = [ record for record in records if record["reachability"] in {"public_dns", "internet_facing_lb"} ] gateway_routed_records = [ record for record in records if record["reachability"] == "gateway_routed_unknown_host" ] internet_reachable_counts = Counter(record["deployment_object"] for record in internet_reachable_records) gateway_routed_counts = Counter(record["deployment_object"] for record in gateway_routed_records) summary = { "record_count": len(records), "deployment_object_count": len({record["deployment_object"] for record in records}), "source_repo_count": len({record["source_repo"] for record in records if record["source_repo"]}), "exposure_surface_counts": dict(exposure_surface_counts), "reachability_counts": dict(reachability_counts), "internet_reachable_record_count": len(internet_reachable_records), "gateway_routed_unknown_host_count": len(gateway_routed_records), "top_objects_by_entrypoints": deployment_object_counts.most_common(20), "top_objects_by_internet_reachable_entrypoints": internet_reachable_counts.most_common(20), "top_objects_by_gateway_routed_entrypoints": gateway_routed_counts.most_common(20), } catalog = { "date": "2026-03-30", "generator": "construction/scan-public-entrypoints.py", "purpose": "deployment-side scan of externally reachable or likely-public entrypoints from app-delivery manifests", "source_paths": [ "app-delivery/**", "knowledge/deployment/app-delivery-object-mapping.json", "knowledge/repos/*/interfaces/*.json", ], "scope_notes": [ "first-pass deployment-side scan only", "focuses on Ingress, HTTPRoute, GRPCRoute, Gateway, and Service(type=LoadBalancer|NodePort)", "does not yet fully resolve shared gateway host ownership", "interface_artifacts are repo-level lookup hints, not exact backend-binary matching", ], "summary": summary, "records": sorted(records, key=lambda item: (item["deployment_object"], item["source_file"], item["entrypoint_id"])), } OUTPUT_JSON.write_text(json.dumps(catalog, indent=2, ensure_ascii=False) + "\n") lines: list[str] = [] lines.append("# Public Entrypoint Scan") lines.append("") lines.append("Date: 2026-03-30") lines.append("") lines.append("## Purpose") lines.append("") lines.append("This document summarizes the first deployment-side scan for public or externally reachable entrypoints.") lines.append("") lines.append("It is intended to answer which deployment objects appear to expose user-facing or externally reachable surfaces, and where their interface artifacts likely live.") lines.append("") lines.append("## Summary") lines.append("") lines.append(f"- entrypoint records: `{summary['record_count']}`") lines.append(f"- deployment objects with entrypoints: `{summary['deployment_object_count']}`") lines.append(f"- mapped source repos represented: `{summary['source_repo_count']}`") lines.append(f"- internet-reachable records: `{summary['internet_reachable_record_count']}`") lines.append(f"- gateway-routed records with unknown host ownership: `{summary['gateway_routed_unknown_host_count']}`") lines.append("") lines.append("Exposure surface counts:") for key, value in exposure_surface_counts.most_common(): lines.append(f"- `{key}`: `{value}`") lines.append("") lines.append("Reachability counts:") for key, value in reachability_counts.most_common(): lines.append(f"- `{key}`: `{value}`") lines.append("") lines.append("## Highest-Signal Internet-Reachable Candidates") lines.append("") for deployment_object, count in internet_reachable_counts.most_common(15): sample = next(record for record in internet_reachable_records if record["deployment_object"] == deployment_object) lines.append( f"- `{deployment_object}`: `{count}` internet-reachable records, source owner `{sample.get('source_repo')}`, sample host `{sample.get('host')}`, sample file `{sample.get('source_file')}`" ) lines.append("") lines.append("## Shared-Gateway Candidates") lines.append("") for deployment_object, count in gateway_routed_counts.most_common(10): sample = next(record for record in gateway_routed_records if record["deployment_object"] == deployment_object) lines.append( f"- `{deployment_object}`: `{count}` gateway-routed records, source owner `{sample.get('source_repo')}`, sample file `{sample.get('source_file')}`" ) lines.append("") lines.append("## Notes") lines.append("") lines.append("- `public_dns` means the manifest exposes an explicit hostname that looks externally meaningful, often under `tidbcloud.com` or `tidbapi.com`.") lines.append("- `gateway_routed_unknown_host` means the object is attached to shared gateway infrastructure but the local manifest does not carry the final hostname.") lines.append("- `internal_lb` means a load balancer is exposed at the network layer but marked internal-only in deployment annotations.") lines.append("- Repo interface artifacts are lookup hints and may include more than the exact backend binary serving that entrypoint.") lines.append("") lines.append("## Machine-Readable Output") lines.append("") lines.append("- `knowledge/deployment/public-entrypoint-catalog.json`") lines.append("") OUTPUT_MD.write_text("\n".join(lines) + "\n") if __name__ == "__main__": main()