Initial release: airndb
This commit is contained in:
admin
271
scripts/airndb_mode.py
Normal file
271
scripts/airndb_mode.py
Normal file
@@ -0,0 +1,271 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Bootstrap AirNDB network debugging context files."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Dict, Tuple
|
||||
|
||||
from airndb_capture import ensure_capture_tool
|
||||
|
||||
MARKER_BEGIN = "<!-- AIRNDB:BEGIN -->"
|
||||
MARKER_END = "<!-- AIRNDB:END -->"
|
||||
|
||||
|
||||
def now_iso() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def airndb_agents_block() -> str:
|
||||
return f"""{MARKER_BEGIN}
|
||||
## AirNDB Network Debug Workflow
|
||||
|
||||
1. Use AirNDB for `/airndb` sessions that need tcpdump/WinDump packet capture, pcap reading, BPF filters, or network evidence.
|
||||
2. On first entry, detect tcpdump/WinDump; on Windows, if none is available, download official `WinDump.exe` from https://www.winpcap.org/windump/install/, verify SHA1, and write `AirPlan/state/airndb/tool.env`.
|
||||
3. Only capture authorized traffic; prefer short bounded captures with `-nn`, `-s 0`, `-c <count>`, and narrow BPF filters.
|
||||
4. Before capture, load:
|
||||
- `AirPlan/AGENTS.md`
|
||||
- `AirPlan/docs/architecture/adr/`
|
||||
- `AirPlan/docs/architecture/c4/module.md`
|
||||
- `AirPlan/docs/network/airndb-log.md`
|
||||
5. Store pcap, text summaries, and JSON reports under `AirPlan/docs/network/airndb-captures/`.
|
||||
6. Record exact command, interface, filter, capture window, artifact paths, key observations, and residual risk in `AirPlan/docs/network/airndb-log.md`.
|
||||
7. AirDbg may call AirNDB when debugging needs DNS/TCP/UDP/TLS/HTTP, ports, proxy, firewall, packet-loss, retransmit, reset, or pcap evidence.
|
||||
8. Update ADR/C4 when network boundaries, capture tooling, observability, ports, protocols, DNS, proxy, TLS, or runtime topology become durable architecture context.
|
||||
{MARKER_END}
|
||||
"""
|
||||
|
||||
|
||||
def c4_module_template() -> str:
|
||||
return """# C4 Module
|
||||
|
||||
## System Context
|
||||
- TODO: Describe the system, users, and important external systems.
|
||||
|
||||
## Containers
|
||||
- TODO: Describe runtime/deployable units and network boundaries.
|
||||
|
||||
## Modules
|
||||
|
||||
| Module | Responsibility | Network Interfaces | Dependencies | Data Ownership | Network Debug Notes |
|
||||
| --- | --- | --- | --- | --- | --- |
|
||||
| TODO | TODO | TODO | TODO | TODO | TODO |
|
||||
|
||||
## Network / Observability Boundaries
|
||||
- tcpdump/WinDump capture points, interfaces, container/WSL/VM/host boundaries: TODO
|
||||
- Ports, protocols, DNS, proxy, TLS, firewall, NAT, or gateway notes: TODO
|
||||
|
||||
## Change Log
|
||||
- TODO: Record network-boundary or observability changes discovered by AirNDB.
|
||||
"""
|
||||
|
||||
|
||||
def adr_template() -> str:
|
||||
return """# ADR-0001: AirNDB Packet Capture Governance
|
||||
|
||||
- Status: Accepted
|
||||
- Date: TODO
|
||||
|
||||
## Context
|
||||
Network debugging needs durable, AI-readable packet capture context and bounded evidence collection.
|
||||
|
||||
## Decision
|
||||
Use AirNDB to build safe tcpdump/WinDump commands, capture or read pcap artifacts, and maintain `AirPlan/docs/network/airndb-log.md`, C4 module docs, and ADR records when network boundaries or diagnostics change.
|
||||
|
||||
## Consequences
|
||||
- Packet evidence can be reused by future AirDbg or AirNDB sessions.
|
||||
- Captures must stay bounded and authorized.
|
||||
- pcap artifacts may contain sensitive data and need careful handling.
|
||||
|
||||
## Alternatives
|
||||
- Chat-only packet notes: rejected because commands, filters, and pcap paths are easy to lose.
|
||||
"""
|
||||
|
||||
|
||||
def network_log_template() -> str:
|
||||
return """# AirNDB Network Debug Log
|
||||
|
||||
Append entries for AirNDB packet capture sessions.
|
||||
|
||||
## Entry Template
|
||||
|
||||
### YYYY-MM-DD: short network issue title
|
||||
|
||||
- Scope / authorization: TODO
|
||||
- Symptom: TODO
|
||||
- Interface: TODO
|
||||
- Filter: TODO
|
||||
- Capture window: TODO
|
||||
- Command: TODO
|
||||
- Artifacts: TODO
|
||||
- Key observations: TODO
|
||||
- AirDbg handoff: TODO
|
||||
- ADR/C4 updates: TODO
|
||||
- Residual risk: TODO
|
||||
"""
|
||||
|
||||
|
||||
def captures_gitignore_template() -> str:
|
||||
return """*.pcap
|
||||
*.pcapng
|
||||
*.cap
|
||||
*.txt
|
||||
*.json
|
||||
!.gitignore
|
||||
"""
|
||||
|
||||
|
||||
def airndb_gitignore_template() -> str:
|
||||
return """tool.env
|
||||
"""
|
||||
|
||||
|
||||
def write_if_missing(path: Path, content: str) -> bool:
|
||||
if path.exists():
|
||||
return False
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(content, encoding="utf-8", newline="\n")
|
||||
return True
|
||||
|
||||
|
||||
def upsert_agents_md(path: Path) -> str:
|
||||
block = airndb_agents_block().rstrip() + "\n"
|
||||
if path.exists():
|
||||
original = path.read_text(encoding="utf-8")
|
||||
existed = True
|
||||
else:
|
||||
original = "# AGENTS.md\n\n"
|
||||
existed = False
|
||||
|
||||
begin = original.find(MARKER_BEGIN)
|
||||
end = original.find(MARKER_END)
|
||||
if begin >= 0 and end > begin:
|
||||
end += len(MARKER_END)
|
||||
updated = original[:begin].rstrip() + "\n\n" + block + original[end:].lstrip()
|
||||
status = "updated"
|
||||
else:
|
||||
updated = original.rstrip() + "\n\n" + block
|
||||
status = "updated" if existed else "created"
|
||||
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(updated, encoding="utf-8", newline="\n")
|
||||
return status
|
||||
|
||||
|
||||
def artifact_map(project_root: Path) -> Dict[str, Path]:
|
||||
return {
|
||||
"AGENTS.md": project_root / "AirPlan" / "AGENTS.md",
|
||||
"c4_module": project_root / "AirPlan" / "docs" / "architecture" / "c4" / "module.md",
|
||||
"adr_dir": project_root / "AirPlan" / "docs" / "architecture" / "adr",
|
||||
"adr_0001": project_root / "AirPlan" / "docs" / "architecture" / "adr" / "ADR-0001-airndb-packet-capture-governance.md",
|
||||
"network_log": project_root / "AirPlan" / "docs" / "network" / "airndb-log.md",
|
||||
"captures_dir": project_root / "AirPlan" / "docs" / "network" / "airndb-captures",
|
||||
"captures_gitignore": project_root / "AirPlan" / "docs" / "network" / "airndb-captures" / ".gitignore",
|
||||
"tool_env": project_root / "AirPlan" / "state" / "airndb" / "tool.env",
|
||||
"airndb_gitignore": project_root / "AirPlan" / "state" / "airndb" / ".gitignore",
|
||||
"state": project_root / "AirPlan" / "state" / "airndb" / "state.json",
|
||||
}
|
||||
|
||||
|
||||
def write_state(path: Path, enabled: bool, project_root: Path) -> None:
|
||||
artifacts = artifact_map(project_root)
|
||||
health = {
|
||||
name: artifacts[name].exists()
|
||||
for name in [
|
||||
"AGENTS.md",
|
||||
"c4_module",
|
||||
"adr_dir",
|
||||
"adr_0001",
|
||||
"network_log",
|
||||
"captures_dir",
|
||||
"captures_gitignore",
|
||||
"tool_env",
|
||||
"airndb_gitignore",
|
||||
]
|
||||
}
|
||||
payload = {
|
||||
"enabled": enabled,
|
||||
"updatedAt": now_iso(),
|
||||
"projectRoot": str(project_root),
|
||||
"artifactHealth": health,
|
||||
"captureTool": ensure_capture_tool(project_root, auto_install=False),
|
||||
}
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def enter_mode(project_root: Path) -> Tuple[str, Dict[str, str]]:
|
||||
artifacts = artifact_map(project_root)
|
||||
results: Dict[str, str] = {}
|
||||
results["AGENTS.md"] = upsert_agents_md(artifacts["AGENTS.md"])
|
||||
results["c4_module"] = "created" if write_if_missing(artifacts["c4_module"], c4_module_template()) else "exists"
|
||||
artifacts["adr_dir"].mkdir(parents=True, exist_ok=True)
|
||||
results["adr_dir"] = "exists"
|
||||
results["adr_0001"] = "created" if write_if_missing(artifacts["adr_0001"], adr_template()) else "exists"
|
||||
results["network_log"] = "created" if write_if_missing(artifacts["network_log"], network_log_template()) else "exists"
|
||||
artifacts["captures_dir"].mkdir(parents=True, exist_ok=True)
|
||||
results["captures_dir"] = "exists"
|
||||
results["captures_gitignore"] = "created" if write_if_missing(artifacts["captures_gitignore"], captures_gitignore_template()) else "exists"
|
||||
results["airndb_gitignore"] = "created" if write_if_missing(artifacts["airndb_gitignore"], airndb_gitignore_template()) else "exists"
|
||||
tool_status = ensure_capture_tool(project_root, auto_install=True)
|
||||
results["capture_tool"] = tool_status.get("status", "unknown")
|
||||
results["capture_tool_path"] = tool_status.get("tool", "")
|
||||
results["capture_tool_source"] = tool_status.get("source", "")
|
||||
results["capture_tool_env"] = tool_status.get("projectEnv", "")
|
||||
if tool_status.get("driverHint"):
|
||||
results["capture_driver_hint"] = tool_status["driverHint"]
|
||||
write_state(artifacts["state"], True, project_root)
|
||||
return "enabled", results
|
||||
|
||||
|
||||
def exit_mode(project_root: Path) -> Tuple[str, Dict[str, str]]:
|
||||
artifacts = artifact_map(project_root)
|
||||
write_state(artifacts["state"], False, project_root)
|
||||
return "disabled", {}
|
||||
|
||||
|
||||
def status_mode(project_root: Path) -> Tuple[str, Dict[str, str]]:
|
||||
artifacts = artifact_map(project_root)
|
||||
state_file = artifacts["state"]
|
||||
enabled = False
|
||||
if state_file.exists():
|
||||
try:
|
||||
payload = json.loads(state_file.read_text(encoding="utf-8"))
|
||||
enabled = bool(payload.get("enabled"))
|
||||
except json.JSONDecodeError:
|
||||
enabled = False
|
||||
results = {
|
||||
name: ("ok" if path.exists() else "missing")
|
||||
for name, path in artifacts.items()
|
||||
if name != "state"
|
||||
}
|
||||
return ("enabled" if enabled else "disabled"), results
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="Manage AirNDB network debug artifacts.")
|
||||
parser.add_argument("--mode", choices=["enter", "setup", "exit", "status"], default="enter")
|
||||
parser.add_argument("--project", default=".")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
project_root = Path(args.project).expanduser().resolve()
|
||||
if args.mode in {"enter", "setup"}:
|
||||
mode_state, result = enter_mode(project_root)
|
||||
elif args.mode == "exit":
|
||||
mode_state, result = exit_mode(project_root)
|
||||
else:
|
||||
mode_state, result = status_mode(project_root)
|
||||
print(f"airndb_mode={mode_state}")
|
||||
print(f"project_root={project_root}")
|
||||
for key, value in result.items():
|
||||
print(f"{key}={value}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user