Edge AI Security Atlas

#!/usr/bin/env python3
"""
A5 local-only simulator: direct backend exposure.

This program evaluates synthetic reachability decisions for an owned Edge AI
lab topology. It does not open sockets, scan ports, sniff traffic, or contact
Jetson/Zynq/Pi devices. Use it to reason about D4 policy before any real lab
validation.
"""

import argparse
from dataclasses import dataclass
from typing import Dict, List, Tuple


@dataclass(frozen=True)
class Host:
    name: str
    vlan: str
    ip: str


@dataclass(frozen=True)
class Service:
    host: str
    port: int
    bind: str
    allowed_sources: Tuple[str, ...]


HOSTS: Dict[str, Host] = {
    "client": Host("client", "client_vlan", "192.168.10.100"),
    "pi_lan1": Host("pi_lan1", "client_vlan", "192.168.10.2"),
    "pi_lan2": Host("pi_lan2", "private_lan2", "192.168.30.1"),
    "jetson": Host("jetson", "private_lan2", "192.168.30.10"),
    "zynq": Host("zynq", "private_lan2", "192.168.30.20"),
}

REQUIRED = [("pi_lan2", "jetson", 8001), ("pi_lan2", "zynq", 8002)]
FORBIDDEN = [("client", "jetson", 8001), ("client", "zynq", 8002)]


def build_services(mode: str) -> Dict[Tuple[str, int], Service]:
    if mode == "exposed":
        return {
            ("jetson", 8001): Service("jetson", 8001, "0.0.0.0", ("client", "pi_lan2")),
            ("zynq", 8002): Service("zynq", 8002, "0.0.0.0", ("client", "pi_lan2")),
        }
    return {
        ("jetson", 8001): Service("jetson", 8001, "192.168.30.10", ("pi_lan2",)),
        ("zynq", 8002): Service("zynq", 8002, "192.168.30.20", ("pi_lan2",)),
    }


def vlan_allows(src: str, dst: str, mode: str) -> bool:
    if src == "pi_lan2" and HOSTS[dst].vlan == "private_lan2":
        return True
    if mode == "exposed":
        return True
    return HOSTS[src].vlan == HOSTS[dst].vlan


def route_exists(src: str, dst: str, mode: str) -> bool:
    if mode == "exposed":
        return True
    return src == "pi_lan2" and HOSTS[dst].vlan == "private_lan2"


def service_bound(service: Service, dst: str) -> bool:
    return service.bind == "0.0.0.0" or service.bind == HOSTS[dst].ip


def firewall_allows(service: Service, src: str) -> bool:
    return src in service.allowed_sources


def reachability(src: str, dst: str, port: int, mode: str) -> Tuple[bool, List[str]]:
    services = build_services(mode)
    service = services.get((dst, port))
    reasons = []
    if not service:
        return False, ["no_service"]
    checks = {
        "route": route_exists(src, dst, mode),
        "vlan": vlan_allows(src, dst, mode),
        "service_bind": service_bound(service, dst),
        "host_firewall": firewall_allows(service, src),
    }
    for name, ok in checks.items():
        if not ok:
            reasons.append("blocked_by_" + name)
    return all(checks.values()), reasons or ["allowed"]


def evaluate(mode: str) -> int:
    exposure_score = 0
    print("mode=%s" % mode)
    for src, dst, port in FORBIDDEN:
        allowed, reasons = reachability(src, dst, port, mode)
        if allowed:
            exposure_score += 1
        print(
            "tuple=forbidden src=%s dst=%s port=%s decision=%s reasons=%s"
            % (src, dst, port, "ALLOW" if allowed else "DROP", ",".join(reasons))
        )
    for src, dst, port in REQUIRED:
        allowed, reasons = reachability(src, dst, port, mode)
        print(
            "tuple=required src=%s dst=%s port=%s decision=%s reasons=%s"
            % (src, dst, port, "ALLOW" if allowed else "DROP", ",".join(reasons))
        )
    print("exposure_score=%s" % exposure_score)
    return exposure_score


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=["exposed", "isolated"], default="exposed")
    args = parser.parse_args()
    score = evaluate(args.mode)
    if args.mode == "isolated" and score == 0:
        print("result=pass d4_private_backend_subnet_enforced")
    elif score > 0:
        print("result=fail direct_backend_exposure_present")
    else:
        print("result=check_required_paths")


if __name__ == "__main__":
    main()

# Vulnerable model: client can reach backend services directly in the synthetic policy.
python3 a5_backend_exposure_policy_lab.py --mode exposed

# Secure model: only Pi LAN2 can reach Jetson :8001 and Zynq :8002.
python3 a5_backend_exposure_policy_lab.py --mode isolated

# Expected secure result:
# tuple=forbidden src=client dst=jetson port=8001 decision=DROP ...
# tuple=forbidden src=client dst=zynq port=8002 decision=DROP ...
# result=pass d4_private_backend_subnet_enforced