diff --git a/.agents/skills/network-developer-interfaces/SKILL.md b/.agents/skills/network-developer-interfaces/SKILL.md index 3709723c..0578bf4c 100644 --- a/.agents/skills/network-developer-interfaces/SKILL.md +++ b/.agents/skills/network-developer-interfaces/SKILL.md @@ -37,6 +37,7 @@ Execute authorized, bounded exposure and configuration assessments across develo - Application-layer findings (e.g. AJP servlet exposure, Docker registry manifest contents) route to web and application specialists (`cops-web-specialist`, `cops-appsec-engineer`). - Container cluster and cloud control-plane interfaces route to cloud infrastructure specialists (`cops-cloud-specialist`). - Host breakout and remote code execution findings route to penetration testing specialists (`cops-pentest-specialist`, `cops-redteam-operator`). + - Before cluster handoff, use `developer-services handoff --kind cluster` with exact destination allowlists and a JSON kubeconfig. The command validates context metadata offline and reports `connectivity_verified: false`. 4. **Crucial Truth Boundary: Inaccessible != Secure**: - If an interface times out, is connection-refused, or is filtered by network controls, it is strictly recorded as `inaccessible` with `auth_prerequisite: unknown` and explicit uncertainty notes. @@ -65,7 +66,9 @@ Execute authorized, bounded exposure and configuration assessments across develo ### 1. Assess Developer and Runtime Interface Services ```bash python3 -m cops developer-services assess \ - --targets "198.51.100.50,dev-runner01.corp.internal" \ + --mode live \ + --targets "dev-runner01.corp.internal" \ + --destination-policy destination-policy.json \ --vantage internal \ --canary-id "canary_debug_probe" \ --output developer_assessment.json @@ -74,7 +77,9 @@ python3 -m cops developer-services assess \ With authorized non-destructive code execution verification: ```bash python3 -m cops developer-services assess \ - --targets "198.51.100.50" \ + --mode live \ + --targets "dev-runner01.corp.internal" \ + --destination-policy destination-policy.json \ --services "docker,jdwp,distcc" \ --allow-code-execution \ --canary-id "canary_docker_probe" \ @@ -84,10 +89,25 @@ python3 -m cops developer-services assess \ Or via the discovery namespace: ```bash python3 -m cops discovery developer assess \ - --targets "198.51.100.50" \ + --mode live \ + --targets "dev-runner01.corp.internal" \ + --destination-policy destination-policy.json \ --output developer_assessment.json ``` +The policy JSON lists exact `hosts` and `resolved_ips` for live collection, plus any approved `repositories`, `images`, `runtimes`, and `clusters` for handoff. See the package README for the schema and example. The collector checks the host before DNS and the resolved IP before socket use. + +### Prepare a Cluster Handoff +```bash +python3 -m cops developer-services handoff \ + --kind cluster --target-host dev-runner01.corp.internal \ + --destination-policy destination-policy.json \ + --cluster lab-cluster --kubeconfig kubeconfig.json \ + --output cluster_handoff.json +``` + +The JSON kubeconfig must have a current context selecting `lab-cluster` and an allowed HTTPS API server hostname. This offline command does not verify connectivity or credentials. + ### 2. Export Developer Privilege Candidates ```bash python3 -m cops developer-services candidates developer_assessment.json \ diff --git a/.claude/skills/network-developer-interfaces/SKILL.md b/.claude/skills/network-developer-interfaces/SKILL.md index 3709723c..0578bf4c 100644 --- a/.claude/skills/network-developer-interfaces/SKILL.md +++ b/.claude/skills/network-developer-interfaces/SKILL.md @@ -37,6 +37,7 @@ Execute authorized, bounded exposure and configuration assessments across develo - Application-layer findings (e.g. AJP servlet exposure, Docker registry manifest contents) route to web and application specialists (`cops-web-specialist`, `cops-appsec-engineer`). - Container cluster and cloud control-plane interfaces route to cloud infrastructure specialists (`cops-cloud-specialist`). - Host breakout and remote code execution findings route to penetration testing specialists (`cops-pentest-specialist`, `cops-redteam-operator`). + - Before cluster handoff, use `developer-services handoff --kind cluster` with exact destination allowlists and a JSON kubeconfig. The command validates context metadata offline and reports `connectivity_verified: false`. 4. **Crucial Truth Boundary: Inaccessible != Secure**: - If an interface times out, is connection-refused, or is filtered by network controls, it is strictly recorded as `inaccessible` with `auth_prerequisite: unknown` and explicit uncertainty notes. @@ -65,7 +66,9 @@ Execute authorized, bounded exposure and configuration assessments across develo ### 1. Assess Developer and Runtime Interface Services ```bash python3 -m cops developer-services assess \ - --targets "198.51.100.50,dev-runner01.corp.internal" \ + --mode live \ + --targets "dev-runner01.corp.internal" \ + --destination-policy destination-policy.json \ --vantage internal \ --canary-id "canary_debug_probe" \ --output developer_assessment.json @@ -74,7 +77,9 @@ python3 -m cops developer-services assess \ With authorized non-destructive code execution verification: ```bash python3 -m cops developer-services assess \ - --targets "198.51.100.50" \ + --mode live \ + --targets "dev-runner01.corp.internal" \ + --destination-policy destination-policy.json \ --services "docker,jdwp,distcc" \ --allow-code-execution \ --canary-id "canary_docker_probe" \ @@ -84,10 +89,25 @@ python3 -m cops developer-services assess \ Or via the discovery namespace: ```bash python3 -m cops discovery developer assess \ - --targets "198.51.100.50" \ + --mode live \ + --targets "dev-runner01.corp.internal" \ + --destination-policy destination-policy.json \ --output developer_assessment.json ``` +The policy JSON lists exact `hosts` and `resolved_ips` for live collection, plus any approved `repositories`, `images`, `runtimes`, and `clusters` for handoff. See the package README for the schema and example. The collector checks the host before DNS and the resolved IP before socket use. + +### Prepare a Cluster Handoff +```bash +python3 -m cops developer-services handoff \ + --kind cluster --target-host dev-runner01.corp.internal \ + --destination-policy destination-policy.json \ + --cluster lab-cluster --kubeconfig kubeconfig.json \ + --output cluster_handoff.json +``` + +The JSON kubeconfig must have a current context selecting `lab-cluster` and an allowed HTTPS API server hostname. This offline command does not verify connectivity or credentials. + ### 2. Export Developer Privilege Candidates ```bash python3 -m cops developer-services candidates developer_assessment.json \ diff --git a/cops/discovery/__init__.py b/cops/discovery/__init__.py index 0ff1562e..6f9bd281 100644 --- a/cops/discovery/__init__.py +++ b/cops/discovery/__init__.py @@ -51,6 +51,11 @@ StandardSocketDeveloperCollector, assess_developer_services, ) +from .developer_destinations import ( + DestinationScopeError, + DeveloperDestinationPolicy, + prepare_developer_handoff, +) from .developer_models import ( DeveloperAuthPrerequisite, DeveloperCategory, @@ -253,10 +258,13 @@ "DeveloperServicesReport", "DeveloperServiceType", "DeveloperServicesCollector", + "DeveloperDestinationPolicy", + "DestinationScopeError", "ExecutionEffect", "OfflineSyntheticDeveloperCollector", "StandardSocketDeveloperCollector", "assess_developer_services", + "prepare_developer_handoff", "DEFAULT_LEGACY_PORTS", "LegacyAuthPrerequisite", "LegacyCategory", diff --git a/cops/discovery/cli.py b/cops/discovery/cli.py index fe70bc05..6d13df86 100644 --- a/cops/discovery/cli.py +++ b/cops/discovery/cli.py @@ -32,6 +32,11 @@ StandardSocketDeveloperCollector, assess_developer_services, ) +from .developer_destinations import ( + DestinationScopeError, + DeveloperDestinationPolicy, + prepare_developer_handoff, +) from .developer_models import ( DeveloperServicesReport, ) @@ -90,18 +95,32 @@ def build_active_parser(parser: argparse.ArgumentParser) -> None: plan_p = active_subs.add_parser("plan", help="Create an active scanning session plan with budgets and targets.") plan_p.add_argument("--targets", "-t", required=True, help="Comma-separated target hosts/IPs or path to JSON.") plan_p.add_argument("--ports", "-p", default="80,443,22", help="Comma-separated ports (default: 80,443,22).") - plan_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Scan vantage point.") + plan_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Scan vantage point.", + ) plan_p.add_argument("--rate-limit", type=float, default=10.0, help="Max probes per second (default: 10.0).") plan_p.add_argument("--timeout", type=float, default=2.0, help="Per-probe timeout in seconds (default: 2.0).") - plan_p.add_argument("--max-total-seconds", type=float, default=None, help="Maximum total execution budget in seconds.") + plan_p.add_argument( + "--max-total-seconds", type=float, default=None, help="Maximum total execution budget in seconds." + ) plan_p.add_argument("--scope-ref", default="authorized-scope", help="Reference identifier of authorized scope.") plan_p.add_argument("--output", "-o", default=None, help="Output path for session JSON.") # 2. scan scan_p = active_subs.add_parser("scan", help="Run bounded active assessment against planned session.") scan_p.add_argument("session", help="Path to ActiveScanSession plan JSON.") - scan_p.add_argument("--scope", "-s", default=None, help="Path to scope boundaries JSON (for DNS rebind/exclusion enforcement).") - scan_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Execution mode (default: synthetic offline).") + scan_p.add_argument( + "--scope", "-s", default=None, help="Path to scope boundaries JSON (for DNS rebind/exclusion enforcement)." + ) + scan_p.add_argument( + "--mode", + choices=["synthetic", "live"], + default="synthetic", + help="Execution mode (default: synthetic offline).", + ) scan_p.add_argument("--offline-targets", default=None, help="Path to mock targets JSON for synthetic dispatcher.") scan_p.add_argument("--checkpoint", "-c", default=None, help="Path to save execution checkpoint.") scan_p.add_argument("--output", "-o", default=None, help="Output path for completed session JSON.") @@ -116,7 +135,9 @@ def build_active_parser(parser: argparse.ArgumentParser) -> None: res_p.add_argument("--output", "-o", default=None, help="Output path for completed session JSON.") # 4. diff - diff_p = active_subs.add_parser("diff", help="Compare baseline and current active scan sessions for remediated exposures.") + diff_p = active_subs.add_parser( + "diff", help="Compare baseline and current active scan sessions for remediated exposures." + ) diff_p.add_argument("baseline", help="Path to baseline session JSON.") diff_p.add_argument("current", help="Path to current/re-test session JSON.") diff_p.add_argument("--output", "-o", default=None, help="Output path for scan delta JSON.") @@ -137,8 +158,18 @@ def build_infra_parser(parser: argparse.ArgumentParser) -> None: # 1. assess ass_p = infra_subs.add_parser("assess", help="Assess infrastructure and identity-facing services.") ass_p.add_argument("--targets", "-t", required=True, help="Comma-separated targets or path to JSON.") - ass_p.add_argument("--services", "-s", default=None, help="Comma-separated service types (dns,mdns,snmp,ntp,rpc,ldap,kerberos,discovery).") - ass_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Probe vantage.") + ass_p.add_argument( + "--services", + "-s", + default=None, + help="Comma-separated service types (dns,mdns,snmp,ntp,rpc,ldap,kerberos,discovery).", + ) + ass_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Probe vantage.", + ) ass_p.add_argument("--scope-ref", default="authorized-scope", help="Scope reference.") ass_p.add_argument("--canary-id", default=None, help="Canary record / controlled identity identifier.") ass_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Collector mode.") @@ -146,7 +177,9 @@ def build_infra_parser(parser: argparse.ArgumentParser) -> None: ass_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 2. candidates - cand_p = infra_subs.add_parser("candidates", help="Export identity attack-path candidates for AD inventory handoff.") + cand_p = infra_subs.add_parser( + "candidates", help="Export identity attack-path candidates for AD inventory handoff." + ) cand_p.add_argument("report", help="Path to InfraAssessmentReport JSON.") cand_p.add_argument("--output", "-o", default=None, help="Output JSON path.") @@ -163,16 +196,32 @@ def build_remote_parser(parser: argparse.ArgumentParser) -> None: # 1. assess ass_p = remote_subs.add_parser("assess", help="Assess remote administration, file sharing, and printing services.") ass_p.add_argument("--targets", "-t", required=True, help="Comma-separated targets or path to JSON.") - ass_p.add_argument("--services", "-s", default=None, help="Comma-separated service types (ssh,telnet,rdp,vnc,winrm,x11,smb,nfs,ftp,tftp,rsync,afp,lpd,ipp,raw_print).") - ass_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Probe vantage.") + ass_p.add_argument( + "--services", + "-s", + default=None, + help="Comma-separated service types (ssh,telnet,rdp,vnc,winrm,x11,smb,nfs,ftp,tftp,rsync,afp,lpd,ipp,raw_print).", + ) + ass_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Probe vantage.", + ) ass_p.add_argument("--scope-ref", default="authorized-scope", help="Scope reference.") - ass_p.add_argument("--canary-id", default=None, help="Canary record / controlled identity / canary file / canary print job identifier.") + ass_p.add_argument( + "--canary-id", + default=None, + help="Canary record / controlled identity / canary file / canary print job identifier.", + ) ass_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Collector mode.") ass_p.add_argument("--offline-targets", default=None, help="Path to mock targets JSON.") ass_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 2. candidates - cand_p = remote_subs.add_parser("candidates", help="Export host privilege candidates for lateral movement specialists.") + cand_p = remote_subs.add_parser( + "candidates", help="Export host privilege candidates for lateral movement specialists." + ) cand_p.add_argument("report", help="Path to RemoteServicesReport JSON.") cand_p.add_argument("--output", "-o", default=None, help="Output JSON path.") @@ -194,17 +243,34 @@ def build_data_parser(parser: argparse.ArgumentParser) -> None: # 1. assess ass_p = data_subs.add_parser("assess", help="Assess databases, caches, and search/analytics services.") ass_p.add_argument("--targets", "-t", required=True, help="Comma-separated targets or path to JSON.") - ass_p.add_argument("--services", "-s", default=None, help="Comma-separated service types (mysql,postgres,mssql,oracle,mongodb,couchdb,cassandra,redis,memcached,elasticsearch,influxdb,kibana,splunk).") - ass_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Probe vantage.") + ass_p.add_argument( + "--services", + "-s", + default=None, + help="Comma-separated service types (mysql,postgres,mssql,oracle,mongodb,couchdb,cassandra,redis,memcached,elasticsearch,influxdb,kibana,splunk).", + ) + ass_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Probe vantage.", + ) ass_p.add_argument("--scope-ref", default="authorized-scope", help="Scope reference.") ass_p.add_argument("--canary-id", default=None, help="Canary record / query / key / index identifier.") - ass_p.add_argument("--query-budget", type=int, default=5, help="Maximum sample rows/documents to inspect without bulk extraction (default: 5).") + ass_p.add_argument( + "--query-budget", + type=int, + default=5, + help="Maximum sample rows/documents to inspect without bulk extraction (default: 5).", + ) ass_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Collector mode.") ass_p.add_argument("--offline-targets", default=None, help="Path to mock targets JSON.") ass_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 2. candidates - cand_p = data_subs.add_parser("candidates", help="Export data privilege candidates for database takeover or lateral movement workflows.") + cand_p = data_subs.add_parser( + "candidates", help="Export data privilege candidates for database takeover or lateral movement workflows." + ) cand_p.add_argument("report", help="Path to DataServicesReport JSON.") cand_p.add_argument("--output", "-o", default=None, help="Output JSON path.") @@ -214,7 +280,9 @@ def build_data_parser(parser: argparse.ArgumentParser) -> None: clean_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 4. inspect - insp_p = data_subs.add_parser("inspect", help="Inspect and summarize database, cache, and search assessment report.") + insp_p = data_subs.add_parser( + "inspect", help="Inspect and summarize database, cache, and search assessment report." + ) insp_p.add_argument("report", help="Path to DataServicesReport JSON.") insp_p.add_argument("--json", action="store_true", help="Emit JSON output.") @@ -226,17 +294,31 @@ def build_messaging_parser(parser: argparse.ArgumentParser) -> None: # 1. assess ass_p = msg_subs.add_parser("assess", help="Assess mail, chat, and message broker services.") ass_p.add_argument("--targets", "-t", required=True, help="Comma-separated targets or path to JSON.") - ass_p.add_argument("--services", "-s", default=None, help="Comma-separated service types (smtp,pop3,imap,irc,rabbitmq,nats,ibmmq,kafka,mqtt).") - ass_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Probe vantage.") + ass_p.add_argument( + "--services", + "-s", + default=None, + help="Comma-separated service types (smtp,pop3,imap,irc,rabbitmq,nats,ibmmq,kafka,mqtt).", + ) + ass_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Probe vantage.", + ) ass_p.add_argument("--scope-ref", default="authorized-scope", help="Scope reference.") ass_p.add_argument("--canary-id", default=None, help="Canary message / queue / mailbox / topic identifier.") - ass_p.add_argument("--message-budget", type=int, default=5, help="Maximum sample messages to transmit or inspect (default: 5).") + ass_p.add_argument( + "--message-budget", type=int, default=5, help="Maximum sample messages to transmit or inspect (default: 5)." + ) ass_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Collector mode.") ass_p.add_argument("--offline-targets", default=None, help="Path to mock targets JSON.") ass_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 2. candidates - cand_p = msg_subs.add_parser("candidates", help="Export messaging privilege candidates for unauthorized relay or broker takeover.") + cand_p = msg_subs.add_parser( + "candidates", help="Export messaging privilege candidates for unauthorized relay or broker takeover." + ) cand_p.add_argument("report", help="Path to MessagingServicesReport JSON.") cand_p.add_argument("--output", "-o", default=None, help="Output JSON path.") @@ -246,7 +328,9 @@ def build_messaging_parser(parser: argparse.ArgumentParser) -> None: clean_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 4. inspect - insp_p = msg_subs.add_parser("inspect", help="Inspect and summarize mail, chat, and message broker assessment report.") + insp_p = msg_subs.add_parser( + "inspect", help="Inspect and summarize mail, chat, and message broker assessment report." + ) insp_p.add_argument("report", help="Path to MessagingServicesReport JSON.") insp_p.add_argument("--json", action="store_true", help="Emit JSON output.") @@ -258,18 +342,45 @@ def build_developer_parser(parser: argparse.ArgumentParser) -> None: # 1. assess ass_p = dev_subs.add_parser("assess", help="Assess developer and runtime interface services.") ass_p.add_argument("--targets", "-t", required=True, help="Comma-separated targets or path to JSON.") - ass_p.add_argument("--services", "-s", default=None, help="Comma-separated service types (docker,docker_registry,rmi,jdwp,erlang_epmd,adb,distcc,svn,ajp,fastcgi).") - ass_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Probe vantage.") + ass_p.add_argument( + "--services", + "-s", + default=None, + help="Comma-separated service types (docker,docker_registry,rmi,jdwp,erlang_epmd,adb,distcc,svn,ajp,fastcgi).", + ) + ass_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Probe vantage.", + ) ass_p.add_argument("--scope-ref", default="authorized-scope", help="Scope reference.") - ass_p.add_argument("--canary-id", default=None, help="Canary record / container / debug session / code artifact identifier.") - ass_p.add_argument("--allow-code-execution", action="store_true", default=False, help="Explicitly authorize code execution probes bound to the approved plan.") - ass_p.add_argument("--allow-state-change", action="store_true", default=False, help="Explicitly authorize state change probes bound to the approved plan.") + ass_p.add_argument( + "--canary-id", default=None, help="Canary record / container / debug session / code artifact identifier." + ) + ass_p.add_argument( + "--allow-code-execution", + action="store_true", + default=False, + help="Explicitly authorize code execution probes bound to the approved plan.", + ) + ass_p.add_argument( + "--allow-state-change", + action="store_true", + default=False, + help="Explicitly authorize state change probes bound to the approved plan.", + ) ass_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Collector mode.") ass_p.add_argument("--offline-targets", default=None, help="Path to mock targets JSON.") + ass_p.add_argument( + "--destination-policy", default=None, help="Exact destination allowlists JSON (required for live mode)." + ) ass_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 2. candidates - cand_p = dev_subs.add_parser("candidates", help="Export developer privilege candidates for code execution or breakout workflows.") + cand_p = dev_subs.add_parser( + "candidates", help="Export developer privilege candidates for code execution or breakout workflows." + ) cand_p.add_argument("report", help="Path to DeveloperServicesReport JSON.") cand_p.add_argument("--output", "-o", default=None, help="Output JSON path.") @@ -279,10 +390,24 @@ def build_developer_parser(parser: argparse.ArgumentParser) -> None: clean_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 4. inspect - insp_p = dev_subs.add_parser("inspect", help="Inspect and summarize developer and runtime interfaces assessment report.") + insp_p = dev_subs.add_parser( + "inspect", help="Inspect and summarize developer and runtime interfaces assessment report." + ) insp_p.add_argument("report", help="Path to DeveloperServicesReport JSON.") insp_p.add_argument("--json", action="store_true", help="Emit JSON output.") + # 5. handoff + hand_p = dev_subs.add_parser("handoff", help="Prepare an offline, scope-checked specialist handoff.") + hand_p.add_argument("--kind", required=True, choices=["application", "host", "cluster"]) + hand_p.add_argument("--target-host", required=True) + hand_p.add_argument("--destination-policy", required=True, help="Exact destination allowlists JSON.") + hand_p.add_argument("--repository") + hand_p.add_argument("--image") + hand_p.add_argument("--runtime") + hand_p.add_argument("--cluster") + hand_p.add_argument("--kubeconfig", help="Kubernetes kubeconfig exported as JSON; required for cluster handoff.") + hand_p.add_argument("--output", "-o", help="Output JSON path.") + def build_legacy_parser(parser: argparse.ArgumentParser) -> None: """Build sub-commands for legacy enterprise, management, and proxy service assessment.""" @@ -291,18 +416,42 @@ def build_legacy_parser(parser: argparse.ArgumentParser) -> None: # 1. assess ass_p = leg_subs.add_parser("assess", help="Assess legacy enterprise, management, and proxy services.") ass_p.add_argument("--targets", "-t", required=True, help="Comma-separated targets or path to JSON.") - ass_p.add_argument("--services", "-s", default=None, help="Comma-separated service types (ndmp,iscsi,ipmi,cisco_smart_install,tacacs,ike,pptp,socks,squid).") - ass_p.add_argument("--vantage", default="external", choices=["external", "internal", "egress_point", "cloud_tenant"], help="Probe vantage.") + ass_p.add_argument( + "--services", + "-s", + default=None, + help="Comma-separated service types (ndmp,iscsi,ipmi,cisco_smart_install,tacacs,ike,pptp,socks,squid).", + ) + ass_p.add_argument( + "--vantage", + default="external", + choices=["external", "internal", "egress_point", "cloud_tenant"], + help="Probe vantage.", + ) ass_p.add_argument("--scope-ref", default="authorized-scope", help="Scope reference.") - ass_p.add_argument("--canary-id", default=None, help="Canary record / container / debug session / code artifact identifier.") - ass_p.add_argument("--allow-code-execution", action="store_true", default=False, help="Explicitly authorize code execution probes bound to the approved plan.") - ass_p.add_argument("--allow-state-change", action="store_true", default=False, help="Explicitly authorize state change probes bound to the approved plan.") + ass_p.add_argument( + "--canary-id", default=None, help="Canary record / container / debug session / code artifact identifier." + ) + ass_p.add_argument( + "--allow-code-execution", + action="store_true", + default=False, + help="Explicitly authorize code execution probes bound to the approved plan.", + ) + ass_p.add_argument( + "--allow-state-change", + action="store_true", + default=False, + help="Explicitly authorize state change probes bound to the approved plan.", + ) ass_p.add_argument("--mode", choices=["synthetic", "live"], default="synthetic", help="Collector mode.") ass_p.add_argument("--offline-targets", default=None, help="Path to mock targets JSON.") ass_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 2. candidates - cand_p = leg_subs.add_parser("candidates", help="Export legacy privilege candidates for takeover or proxy egress pivoting.") + cand_p = leg_subs.add_parser( + "candidates", help="Export legacy privilege candidates for takeover or proxy egress pivoting." + ) cand_p.add_argument("report", help="Path to LegacyServicesReport JSON.") cand_p.add_argument("--output", "-o", default=None, help="Output JSON path.") @@ -312,7 +461,9 @@ def build_legacy_parser(parser: argparse.ArgumentParser) -> None: clean_p.add_argument("--output", "-o", default=None, help="Output JSON path.") # 4. inspect - insp_p = leg_subs.add_parser("inspect", help="Inspect and summarize legacy enterprise and proxy services assessment report.") + insp_p = leg_subs.add_parser( + "inspect", help="Inspect and summarize legacy enterprise and proxy services assessment report." + ) insp_p.add_argument("report", help="Path to LegacyServicesReport JSON.") insp_p.add_argument("--json", action="store_true", help="Emit JSON output.") @@ -340,7 +491,9 @@ def build_discovery_parser(subparsers: argparse._SubParsersAction[Any]) -> argpa norm_p.add_argument("--output", "-o", default=None, help="Output JSON path (prints to stdout if omitted).") # Passive Subcommand: reconcile - rec_p = disc_subs.add_parser("reconcile", help="Reconcile inventory against approved scope and quarantine uncertain assets.") + rec_p = disc_subs.add_parser( + "reconcile", help="Reconcile inventory against approved scope and quarantine uncertain assets." + ) rec_p.add_argument("inventory", help="Path to discovery inventory JSON.") rec_p.add_argument("--scope", "-s", required=True, help="Path to engagement scope or action plan JSON.") rec_p.add_argument("--output", "-o", default=None, help="Output JSON path (prints to stdout if omitted).") @@ -448,7 +601,9 @@ def command_active_discovery(args: argparse.Namespace, root: Path | None = None) out_json = json.dumps(completed_session.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Active scan complete (status: {completed_session.status}, probes: {completed_session.total_probes_dispatched}) saved to {args.output}") + print( + f"Active scan complete (status: {completed_session.status}, probes: {completed_session.total_probes_dispatched}) saved to {args.output}" + ) else: print(out_json) return 0 @@ -491,7 +646,9 @@ def command_active_discovery(args: argparse.Namespace, root: Path | None = None) out_json = json.dumps(delta.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Scan delta ({len(delta.remediated_exposures)} remediated, {delta.remediation_rate}% rate) saved to {args.output}") + print( + f"Scan delta ({len(delta.remediated_exposures)} remediated, {delta.remediation_rate}% rate) saved to {args.output}" + ) else: print(out_json) return 0 @@ -551,7 +708,9 @@ def command_infra_discovery(args: argparse.Namespace, root: Path | None = None) out_json = json.dumps(report.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Assessed {len(report.services_assessed)} infrastructure services ({len(report.attack_path_candidates)} candidates) saved to {args.output}") + print( + f"Assessed {len(report.services_assessed)} infrastructure services ({len(report.attack_path_candidates)} candidates) saved to {args.output}" + ) else: print(out_json) return 0 @@ -588,9 +747,17 @@ def command_infra_discovery(args: argparse.Namespace, root: Path | None = None) print(f"Attack Path Cand: {report.summary.get('attack_path_candidates', len(report.attack_path_candidates))}") print() for s in report.services_assessed: - status_icon = "!" if s.exposure_status in ("exposed", "misconfigured") else ("✓" if s.exposure_status == "protected" else "?") - cand_marker = f" -> CANDIDATE: {s.attack_path_candidate.attack_path_type}" if s.attack_path_candidate else "" - print(f" [{status_icon}] {s.service_type.upper():<10} {s.target_host}:{s.port:<5} {s.exposure_status:<14} (auth: {s.auth_prerequisite}){cand_marker}") + status_icon = ( + "!" + if s.exposure_status in ("exposed", "misconfigured") + else ("✓" if s.exposure_status == "protected" else "?") + ) + cand_marker = ( + f" -> CANDIDATE: {s.attack_path_candidate.attack_path_type}" if s.attack_path_candidate else "" + ) + print( + f" [{status_icon}] {s.service_type.upper():<10} {s.target_host}:{s.port:<5} {s.exposure_status:<14} (auth: {s.auth_prerequisite}){cand_marker}" + ) if report.summary.get("inaccessible", 0) > 0: print() print("Truth-in-Advertising Notice: Inaccessible services are NOT reported as secure or hardened.") @@ -633,7 +800,9 @@ def command_remote_discovery(args: argparse.Namespace, root: Path | None = None) out_json = json.dumps(report.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Assessed {len(report.services_assessed)} remote/file/print services ({len(report.host_privilege_candidates)} candidates, {len(report.cleanup_receipts)} cleanup receipts) saved to {args.output}") + print( + f"Assessed {len(report.services_assessed)} remote/file/print services ({len(report.host_privilege_candidates)} candidates, {len(report.cleanup_receipts)} cleanup receipts) saved to {args.output}" + ) else: print(out_json) return 0 @@ -680,18 +849,26 @@ def command_remote_discovery(args: argparse.Namespace, root: Path | None = None) print(f"Exposed: {report.summary.get('exposed', 0)}") print(f"Protected: {report.summary.get('protected', 0)}") print(f"Inaccessible: {report.summary.get('inaccessible', 0)}") - print(f"Host Privilege Cand: {report.summary.get('host_privilege_candidates', len(report.host_privilege_candidates))}") + print( + f"Host Privilege Cand: {report.summary.get('host_privilege_candidates', len(report.host_privilege_candidates))}" + ) print(f"Cleanup Receipts: {report.summary.get('cleanup_receipts', len(report.cleanup_receipts))}") print() for s in report.services_assessed: - status_icon = "!" if s.exposure_status in ("exposed", "misconfigured") else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + status_icon = ( + "!" + if s.exposure_status in ("exposed", "misconfigured") + else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + ) cand_types = [c.finding_type for c in s.host_privilege_candidates] cand_marker = f" -> CANDIDATES: {', '.join(cand_types)}" if cand_types else "" cat_label = s.category.upper() if isinstance(s.category, str) else s.category.value.upper() stype = s.service_type if isinstance(s.service_type, str) else s.service_type.value auth_val = s.auth_prerequisite if isinstance(s.auth_prerequisite, str) else s.auth_prerequisite.value exp_val = s.exposure_status if isinstance(s.exposure_status, str) else s.exposure_status.value - print(f" [{status_icon}] [{cat_label:<12}] {stype.upper():<10} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}){cand_marker}") + print( + f" [{status_icon}] [{cat_label:<12}] {stype.upper():<10} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}){cand_marker}" + ) if report.summary.get("inaccessible", 0) > 0: print() print("Truth-in-Advertising Notice: Inaccessible services are NOT reported as secure or hardened.") @@ -735,7 +912,9 @@ def command_data_discovery(args: argparse.Namespace, root: Path | None = None) - out_json = json.dumps(report.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Assessed {len(report.services_assessed)} data services ({len(report.privilege_candidates)} candidates, {len(report.cleanup_receipts)} cleanup receipts) saved to {args.output}") + print( + f"Assessed {len(report.services_assessed)} data services ({len(report.privilege_candidates)} candidates, {len(report.cleanup_receipts)} cleanup receipts) saved to {args.output}" + ) else: print(out_json) return 0 @@ -786,14 +965,20 @@ def command_data_discovery(args: argparse.Namespace, root: Path | None = None) - print(f"Cleanup Receipts: {report.summary.get('cleanup_receipts', len(report.cleanup_receipts))}") print() for s in report.services_assessed: - status_icon = "!" if s.exposure_status in ("exposed", "misconfigured") else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + status_icon = ( + "!" + if s.exposure_status in ("exposed", "misconfigured") + else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + ) cand_types = [c.finding_type for c in s.privilege_candidates] cand_marker = f" -> CANDIDATES: {', '.join(cand_types)}" if cand_types else "" cat_label = s.category.upper() if isinstance(s.category, str) else s.category.value.upper() stype = s.service_type if isinstance(s.service_type, str) else s.service_type.value auth_val = s.auth_prerequisite if isinstance(s.auth_prerequisite, str) else s.auth_prerequisite.value exp_val = s.exposure_status if isinstance(s.exposure_status, str) else s.exposure_status.value - print(f" [{status_icon}] [{cat_label:<15}] {stype.upper():<14} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){cand_marker}") + print( + f" [{status_icon}] [{cat_label:<15}] {stype.upper():<14} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){cand_marker}" + ) if report.summary.get("inaccessible", 0) > 0: print() print("Truth-in-Advertising Notice: Inaccessible services are NOT reported as secure or hardened.") @@ -839,7 +1024,9 @@ def command_messaging_discovery(args: argparse.Namespace, root: Path | None = No out_json = json.dumps(report.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Assessed {len(report.assessments)} messaging services ({report.candidates_count} candidates, {report.cleanup_receipts_count} cleanup receipts) saved to {args.output}") + print( + f"Assessed {len(report.assessments)} messaging services ({report.candidates_count} candidates, {report.cleanup_receipts_count} cleanup receipts) saved to {args.output}" + ) else: print(out_json) return 0 @@ -895,7 +1082,11 @@ def command_messaging_discovery(args: argparse.Namespace, root: Path | None = No print(f"Cleanup Receipts: {report.cleanup_receipts_count}") print() for s in report.assessments: - status_icon = "!" if s.exposure_status in ("exposed", "misconfigured") else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + status_icon = ( + "!" + if s.exposure_status in ("exposed", "misconfigured") + else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + ) cand_types = [c.finding_type for c in s.privilege_candidates] cand_types_str = ", ".join(cand_types) cand_marker = f" -> CANDIDATES: {cand_types_str}" if cand_types else "" @@ -903,7 +1094,9 @@ def command_messaging_discovery(args: argparse.Namespace, root: Path | None = No stype = s.service_type if isinstance(s.service_type, str) else s.service_type.value auth_val = s.auth_prerequisite if isinstance(s.auth_prerequisite, str) else s.auth_prerequisite.value exp_val = s.exposure_status if isinstance(s.exposure_status, str) else s.exposure_status.value - print(f" [{status_icon}] [{cat_label:<25}] {stype.upper():<10} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){cand_marker}") + print( + f" [{status_icon}] [{cat_label:<25}] {stype.upper():<10} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){cand_marker}" + ) if report.total_inaccessible > 0: print() print("Truth-in-Advertising Notice: Inaccessible services are NOT reported as secure or hardened.") @@ -916,6 +1109,37 @@ def command_developer_discovery(args: argparse.Namespace, root: Path | None = No """Handle developer and runtime interface discovery CLI subcommands.""" dev_cmd = getattr(args, "developer_command", None) + if dev_cmd == "handoff": + try: + policy = DeveloperDestinationPolicy.load(args.destination_policy) + kubeconfig = json.loads(Path(args.kubeconfig).read_text(encoding="utf-8")) if args.kubeconfig else None + handoff = prepare_developer_handoff( + kind=args.kind, + target_host=args.target_host, + policy=policy, + repository=args.repository, + image=args.image, + runtime=args.runtime, + cluster=args.cluster, + kubeconfig=kubeconfig, + ) + except OSError: + print("Destination scope error: unable to read destination policy or Kubernetes context", file=sys.stderr) + return 2 + except (DestinationScopeError, ValueError) as err: + print(f"Destination scope error: {err}", file=sys.stderr) + return 2 + out_json = json.dumps(handoff, indent=2) + if args.output: + try: + Path(args.output).write_text(out_json, encoding="utf-8") + except OSError: + print("Destination output error: unable to write handoff record", file=sys.stderr) + return 2 + else: + print(out_json) + return 0 + if dev_cmd == "assess": raw_targets = args.targets if Path(raw_targets).is_file(): @@ -939,26 +1163,44 @@ def command_developer_discovery(args: argparse.Namespace, root: Path | None = No allow_state_change=allow_state, ) else: - collector = StandardSocketDeveloperCollector( + if not getattr(args, "destination_policy", None): + print("Destination scope error: --destination-policy is required for live mode", file=sys.stderr) + return 2 + try: + policy = DeveloperDestinationPolicy.load(args.destination_policy) + collector = StandardSocketDeveloperCollector( + destination_policy=policy, + allow_code_execution=allow_exec, + allow_state_change=allow_state, + ) + except OSError: + print("Destination scope error: unable to read destination policy", file=sys.stderr) + return 2 + except (DestinationScopeError, ValueError) as err: + print(f"Destination scope error: {err}", file=sys.stderr) + return 2 + + try: + report = assess_developer_services( + targets=targets, + service_types=services, + collector=collector, + vantage=args.vantage, + scope_ref=args.scope_ref, + canary_id=args.canary_id, allow_code_execution=allow_exec, allow_state_change=allow_state, ) - - report = assess_developer_services( - targets=targets, - service_types=services, - collector=collector, - vantage=args.vantage, - scope_ref=args.scope_ref, - canary_id=args.canary_id, - allow_code_execution=allow_exec, - allow_state_change=allow_state, - ) + except (DestinationScopeError, ValueError) as err: + print(f"Destination scope error: {err}", file=sys.stderr) + return 2 out_json = json.dumps(report.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Assessed {len(report.assessments)} developer interface services ({report.candidates_count} candidates, {report.cleanup_receipts_count} receipts) saved to {args.output}") + print( + f"Assessed {len(report.assessments)} developer interface services ({report.candidates_count} candidates, {report.cleanup_receipts_count} receipts) saved to {args.output}" + ) else: print(out_json) return 0 @@ -1014,7 +1256,11 @@ def command_developer_discovery(args: argparse.Namespace, root: Path | None = No print(f"Cleanup Receipts: {report.cleanup_receipts_count}") print() for s in report.assessments: - status_icon = "!" if s.exposure_status in ("exposed", "misconfigured") else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + status_icon = ( + "!" + if s.exposure_status in ("exposed", "misconfigured") + else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + ) cand_types = [c.finding_type for c in s.privilege_candidates] cand_types_str = ", ".join(cand_types) cand_marker = f" -> CANDIDATES: {cand_types_str}" if cand_types else "" @@ -1023,7 +1269,9 @@ def command_developer_discovery(args: argparse.Namespace, root: Path | None = No auth_val = s.auth_prerequisite if isinstance(s.auth_prerequisite, str) else s.auth_prerequisite.value exp_val = s.exposure_status if isinstance(s.exposure_status, str) else s.exposure_status.value exec_note = " [CODE-EXEC]" if s.can_execute_code else (" [STATE-CHANGE]" if s.can_change_state else "") - print(f" [{status_icon}] [{cat_label:<32}] {stype.upper():<15} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){exec_note}{cand_marker}") + print( + f" [{status_icon}] [{cat_label:<32}] {stype.upper():<15} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){exec_note}{cand_marker}" + ) if report.total_inaccessible > 0: print() print("Truth-in-Advertising Notice: Inaccessible services are NOT reported as secure or hardened.") @@ -1078,7 +1326,9 @@ def command_legacy_discovery(args: argparse.Namespace, root: Path | None = None) out_json = json.dumps(report.to_dict(), indent=2) if args.output: Path(args.output).write_text(out_json, encoding="utf-8") - print(f"Assessed {len(report.assessments)} legacy services ({report.candidates_count} candidates, {report.cleanup_receipts_count} receipts) saved to {args.output}") + print( + f"Assessed {len(report.assessments)} legacy services ({report.candidates_count} candidates, {report.cleanup_receipts_count} receipts) saved to {args.output}" + ) else: print(out_json) return 0 @@ -1134,7 +1384,11 @@ def command_legacy_discovery(args: argparse.Namespace, root: Path | None = None) print(f"Cleanup Receipts: {report.cleanup_receipts_count}") print() for s in report.assessments: - status_icon = "!" if s.exposure_status in ("exposed", "misconfigured") else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + status_icon = ( + "!" + if s.exposure_status in ("exposed", "misconfigured") + else ("✓" if s.exposure_status in ("protected", "remediated") else "?") + ) cand_types = [c.finding_type for c in s.privilege_candidates] cand_types_str = ", ".join(cand_types) cand_marker = f" -> CANDIDATES: {cand_types_str}" if cand_types else "" @@ -1146,7 +1400,9 @@ def command_legacy_discovery(args: argparse.Namespace, root: Path | None = None) proxy_note = "" if s.proxy_egress_tested: proxy_note = " [EGRESS-RESTRICTED]" if s.proxy_egress_restricted else " [EGRESS-OPEN]" - print(f" [{status_icon}] [{cat_label:<35}] {stype.upper():<20} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){exec_note}{proxy_note}{cand_marker}") + print( + f" [{status_icon}] [{cat_label:<35}] {stype.upper():<20} {s.target_host}:{s.port:<5} {exp_val:<14} (auth: {auth_val}, role: {s.assigned_role}){exec_note}{proxy_note}{cand_marker}" + ) if report.total_inaccessible > 0: print() print("Truth-in-Advertising Notice: Inaccessible services are NOT reported as secure or hardened.") diff --git a/cops/discovery/developer_collector.py b/cops/discovery/developer_collector.py index e9a4e0f5..fa583df2 100644 --- a/cops/discovery/developer_collector.py +++ b/cops/discovery/developer_collector.py @@ -8,6 +8,7 @@ from abc import ABC, abstractmethod from typing import Any +from .developer_destinations import DestinationScopeError, DeveloperDestinationPolicy from .developer_models import ( CleanupReceipt, DeveloperAuthPrerequisite, @@ -25,17 +26,14 @@ # Container & Orchestration Runtimes DeveloperServiceType.DOCKER.value: (2375, "tcp", DeveloperCategory.CONTAINER_ORCHESTRATION_RUNTIME.value), DeveloperServiceType.DOCKER_REGISTRY.value: (5000, "tcp", DeveloperCategory.CONTAINER_ORCHESTRATION_RUNTIME.value), - # Language & Debugging Runtimes DeveloperServiceType.RMI.value: (1099, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value), DeveloperServiceType.JDWP.value: (8000, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value), DeveloperServiceType.ERLANG_EPMD.value: (4369, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value), DeveloperServiceType.ADB.value: (5555, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value), - # Distributed Build & SCM DeveloperServiceType.DISTCC.value: (3632, "tcp", DeveloperCategory.DISTRIBUTED_BUILD_SCM.value), DeveloperServiceType.SVN.value: (3690, "tcp", DeveloperCategory.DISTRIBUTED_BUILD_SCM.value), - # Application Server & Gateway Interfaces DeveloperServiceType.AJP.value: (8009, "tcp", DeveloperCategory.APPLICATION_SERVER_GATEWAY.value), DeveloperServiceType.FASTCGI.value: (9000, "tcp", DeveloperCategory.APPLICATION_SERVER_GATEWAY.value), @@ -64,45 +62,215 @@ def probe_service( ) -> DeveloperServiceAssessment: """Probe and evaluate discrete developer service exposure.""" - def assess_docker(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 2375) -> DeveloperServiceAssessment: + def assess_docker( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 2375, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.DOCKER.value, port, "tcp", DeveloperCategory.CONTAINER_ORCHESTRATION_RUNTIME.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.DOCKER.value, + port, + "tcp", + DeveloperCategory.CONTAINER_ORCHESTRATION_RUNTIME.value, + vantage, + canary_artifact, + timeout, + ) - def assess_docker_registry(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 5000) -> DeveloperServiceAssessment: + def assess_docker_registry( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 5000, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.DOCKER_REGISTRY.value, port, "tcp", DeveloperCategory.CONTAINER_ORCHESTRATION_RUNTIME.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.DOCKER_REGISTRY.value, + port, + "tcp", + DeveloperCategory.CONTAINER_ORCHESTRATION_RUNTIME.value, + vantage, + canary_artifact, + timeout, + ) - def assess_rmi(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 1099) -> DeveloperServiceAssessment: + def assess_rmi( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 1099, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.RMI.value, port, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.RMI.value, + port, + "tcp", + DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, + vantage, + canary_artifact, + timeout, + ) - def assess_jdwp(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 8000) -> DeveloperServiceAssessment: + def assess_jdwp( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 8000, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.JDWP.value, port, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.JDWP.value, + port, + "tcp", + DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, + vantage, + canary_artifact, + timeout, + ) - def assess_erlang_epmd(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 4369) -> DeveloperServiceAssessment: + def assess_erlang_epmd( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 4369, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.ERLANG_EPMD.value, port, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.ERLANG_EPMD.value, + port, + "tcp", + DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, + vantage, + canary_artifact, + timeout, + ) - def assess_adb(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 5555) -> DeveloperServiceAssessment: + def assess_adb( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 5555, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.ADB.value, port, "tcp", DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.ADB.value, + port, + "tcp", + DeveloperCategory.LANGUAGE_DEBUG_RUNTIME.value, + vantage, + canary_artifact, + timeout, + ) - def assess_distcc(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 3632) -> DeveloperServiceAssessment: + def assess_distcc( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 3632, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.DISTCC.value, port, "tcp", DeveloperCategory.DISTRIBUTED_BUILD_SCM.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.DISTCC.value, + port, + "tcp", + DeveloperCategory.DISTRIBUTED_BUILD_SCM.value, + vantage, + canary_artifact, + timeout, + ) - def assess_svn(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 3690) -> DeveloperServiceAssessment: + def assess_svn( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 3690, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.SVN.value, port, "tcp", DeveloperCategory.DISTRIBUTED_BUILD_SCM.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.SVN.value, + port, + "tcp", + DeveloperCategory.DISTRIBUTED_BUILD_SCM.value, + vantage, + canary_artifact, + timeout, + ) - def assess_ajp(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 8009) -> DeveloperServiceAssessment: + def assess_ajp( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 8009, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.AJP.value, port, "tcp", DeveloperCategory.APPLICATION_SERVER_GATEWAY.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.AJP.value, + port, + "tcp", + DeveloperCategory.APPLICATION_SERVER_GATEWAY.value, + vantage, + canary_artifact, + timeout, + ) - def assess_fastcgi(self, target_host: str, canary_artifact: str | None = None, vantage: str = "external", timeout: float = 2.0, port: int = 9000) -> DeveloperServiceAssessment: + def assess_fastcgi( + self, + target_host: str, + canary_artifact: str | None = None, + vantage: str = "external", + timeout: float = 2.0, + port: int = 9000, + ) -> DeveloperServiceAssessment: ip = self.resolve_target(target_host) - return self.probe_service(target_host, ip, DeveloperServiceType.FASTCGI.value, port, "tcp", DeveloperCategory.APPLICATION_SERVER_GATEWAY.value, vantage, canary_artifact, timeout) + return self.probe_service( + target_host, + ip, + DeveloperServiceType.FASTCGI.value, + port, + "tcp", + DeveloperCategory.APPLICATION_SERVER_GATEWAY.value, + vantage, + canary_artifact, + timeout, + ) class OfflineSyntheticDeveloperCollector(DeveloperServicesCollector): @@ -139,16 +307,18 @@ def probe_service( canary_artifact: str | None = None, timeout: float = 2.0, ) -> DeveloperServiceAssessment: - self.probes_recorded.append({ - "target_host": target_host, - "resolved_ip": resolved_ip, - "service_type": service_type, - "category": category, - "port": port, - "protocol": protocol, - "vantage": vantage, - "canary_artifact": canary_artifact, - }) + self.probes_recorded.append( + { + "target_host": target_host, + "resolved_ip": resolved_ip, + "service_type": service_type, + "category": category, + "port": port, + "protocol": protocol, + "vantage": vantage, + "canary_artifact": canary_artifact, + } + ) t_data = self.targets.get(target_host) or self.targets.get(resolved_ip) or {} svc_data = t_data.get(service_type) or t_data.get(f"{service_type}:{port}") @@ -186,10 +356,21 @@ def probe_service( details = dict(svc_data.get("details", {})) for k, v in svc_data.items(): if k not in ( - "details", "vulnerabilities", "versions", "status", "protected", - "inaccessible", "remediated", "state", "auth_prerequisite", - "authentication_required", "role", "role_assigned", "can_execute_code", - "can_change_state", "bound_to_plan", + "details", + "vulnerabilities", + "versions", + "status", + "protected", + "inaccessible", + "remediated", + "state", + "auth_prerequisite", + "authentication_required", + "role", + "role_assigned", + "can_execute_code", + "can_change_state", + "bound_to_plan", ): details.setdefault(k, v) @@ -214,9 +395,7 @@ def probe_service( bound_to_plan=True, applicable_versions=svc_data.get("versions", [details.get("version", "Remediated")]), configuration_details=details, - uncertainty_notes=[ - "Service verified remediated against historical unauthenticated exposure" - ], + uncertainty_notes=["Service verified remediated against historical unauthenticated exposure"], ) # 3. Protected Service Check (Properly authenticated & hardened) @@ -227,11 +406,21 @@ def probe_service( is_protected = True elif service_type == DeveloperServiceType.DOCKER_REGISTRY.value and details.get("auth_required") is True: is_protected = True - elif service_type == DeveloperServiceType.RMI.value and details.get("ssl_enabled") is True and details.get("auth_required") is True: + elif ( + service_type == DeveloperServiceType.RMI.value + and details.get("ssl_enabled") is True + and details.get("auth_required") is True + ): is_protected = True - elif service_type == DeveloperServiceType.JDWP.value and (details.get("bind_localhost") is True or details.get("disabled") is True): + elif service_type == DeveloperServiceType.JDWP.value and ( + details.get("bind_localhost") is True or details.get("disabled") is True + ): is_protected = True - elif service_type == DeveloperServiceType.ERLANG_EPMD.value and details.get("cookie_required") is True and details.get("firewalled") is True: + elif ( + service_type == DeveloperServiceType.ERLANG_EPMD.value + and details.get("cookie_required") is True + and details.get("firewalled") is True + ): is_protected = True elif service_type == DeveloperServiceType.ADB.value and details.get("rsa_key_enforced") is True: is_protected = True @@ -239,20 +428,22 @@ def probe_service( is_protected = True elif service_type == DeveloperServiceType.SVN.value and details.get("anon_access_none") is True: is_protected = True - elif service_type == DeveloperServiceType.AJP.value and details.get("secret_required") is True and details.get("secret_configured") is True: + elif ( + service_type == DeveloperServiceType.AJP.value + and details.get("secret_required") is True + and details.get("secret_configured") is True + ): is_protected = True elif service_type == DeveloperServiceType.FASTCGI.value and details.get("bind_localhost") is True: is_protected = True - canary_active = bool( - canary_artifact - or details.get("canary_verified") - or details.get("canary_probe_received") - ) + canary_active = bool(canary_artifact or details.get("canary_verified") or details.get("canary_probe_received")) receipts: list[CleanupReceipt] = [] if canary_active and canary_artifact: - receipt_id = hashlib.sha256(f"clean:{target_host}:{service_type}:{canary_artifact}".encode()).hexdigest()[:16] + receipt_id = hashlib.sha256(f"clean:{target_host}:{service_type}:{canary_artifact}".encode()).hexdigest()[ + :16 + ] receipts.append( CleanupReceipt( receipt_id=f"rec-{receipt_id}", @@ -294,7 +485,9 @@ def probe_service( auth_req = svc_data.get("authentication_required", False) auth_prereq = svc_data.get( "auth_prerequisite", - DeveloperAuthPrerequisite.NONE.value if not auth_req else DeveloperAuthPrerequisite.DEFAULT_CREDENTIALS.value + DeveloperAuthPrerequisite.NONE.value + if not auth_req + else DeveloperAuthPrerequisite.DEFAULT_CREDENTIALS.value, ) assigned_role = svc_data.get("role", "anonymous" if not auth_req else "developer") status = svc_data.get("status", DeveloperExposureStatus.EXPOSED.value) @@ -308,7 +501,9 @@ def probe_service( # Protocol-specific candidate evaluation if service_type == DeveloperServiceType.DOCKER.value: if not auth_req or details.get("tls_verify") is False: - vulns.append("Docker daemon TCP socket exposed without mutual TLS; root container breakout / RCE permitted") + vulns.append( + "Docker daemon TCP socket exposed without mutual TLS; root container breakout / RCE permitted" + ) c_id = hashlib.sha256(f"docker:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -332,7 +527,9 @@ def probe_service( elif service_type == DeveloperServiceType.DOCKER_REGISTRY.value: if not auth_req or details.get("auth_required") is False: - vulns.append("Docker Registry HTTP API exposed without authentication; container image and secret leakage permitted") + vulns.append( + "Docker Registry HTTP API exposed without authentication; container image and secret leakage permitted" + ) c_id = hashlib.sha256(f"registry:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -355,7 +552,9 @@ def probe_service( elif service_type == DeveloperServiceType.RMI.value: if not auth_req or details.get("ssl_enabled") is False: - vulns.append("Java Remote Method Invocation (RMI) registry exposed without authentication; remote class loading / RCE possible") + vulns.append( + "Java Remote Method Invocation (RMI) registry exposed without authentication; remote class loading / RCE possible" + ) c_id = hashlib.sha256(f"rmi:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -379,7 +578,9 @@ def probe_service( elif service_type == DeveloperServiceType.JDWP.value: if details.get("bind_localhost") is False or not auth_req: - vulns.append("Java Debug Wire Protocol (JDWP) port exposed to network; arbitrary JVM bytecode execution permitted") + vulns.append( + "Java Debug Wire Protocol (JDWP) port exposed to network; arbitrary JVM bytecode execution permitted" + ) c_id = hashlib.sha256(f"jdwp:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -403,7 +604,9 @@ def probe_service( elif service_type == DeveloperServiceType.ERLANG_EPMD.value: if not auth_req or details.get("cookie_required") is False: - vulns.append("Erlang Port Mapper Daemon (EPMD) exposed; unauthenticated node discovery and arbitrary command execution") + vulns.append( + "Erlang Port Mapper Daemon (EPMD) exposed; unauthenticated node discovery and arbitrary command execution" + ) c_id = hashlib.sha256(f"epmd:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -427,7 +630,9 @@ def probe_service( elif service_type == DeveloperServiceType.ADB.value: if not auth_req or details.get("rsa_key_enforced") is False: - vulns.append("Android Debug Bridge (ADB) daemon exposed without RSA authorization; arbitrary shell access and app installation permitted") + vulns.append( + "Android Debug Bridge (ADB) daemon exposed without RSA authorization; arbitrary shell access and app installation permitted" + ) c_id = hashlib.sha256(f"adb:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -451,7 +656,9 @@ def probe_service( elif service_type == DeveloperServiceType.DISTCC.value: if not auth_req or details.get("allow_cidr_enforced") is False: - vulns.append("distcc distributed compiler daemon exposed without host filtering; arbitrary command execution (CVE-2004-2687)") + vulns.append( + "distcc distributed compiler daemon exposed without host filtering; arbitrary command execution (CVE-2004-2687)" + ) c_id = hashlib.sha256(f"distcc:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -475,7 +682,9 @@ def probe_service( elif service_type == DeveloperServiceType.SVN.value: if details.get("anon_access_none") is False or not auth_req: - vulns.append("Subversion repository daemon (svnserve) allows anonymous read access; source code repository exposure") + vulns.append( + "Subversion repository daemon (svnserve) allows anonymous read access; source code repository exposure" + ) c_id = hashlib.sha256(f"svn:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -498,7 +707,9 @@ def probe_service( elif service_type == DeveloperServiceType.AJP.value: if details.get("secret_required") is False or not details.get("secret_configured"): - vulns.append("Apache JServ Protocol (AJP13) exposed without secret; Ghostcat arbitrary file read / RCE (CVE-2020-1938)") + vulns.append( + "Apache JServ Protocol (AJP13) exposed without secret; Ghostcat arbitrary file read / RCE (CVE-2020-1938)" + ) c_id = hashlib.sha256(f"ajp:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -521,7 +732,9 @@ def probe_service( elif service_type == DeveloperServiceType.FASTCGI.value: if details.get("bind_localhost") is False or not auth_req: - vulns.append("FastCGI (php-fpm) port exposed to external network; arbitrary PHP code execution / file inclusion") + vulns.append( + "FastCGI (php-fpm) port exposed to external network; arbitrary PHP code execution / file inclusion" + ) c_id = hashlib.sha256(f"fastcgi:{target_host}:{port}".encode()).hexdigest()[:16] candidates.append( DeveloperPrivilegeCandidate( @@ -582,14 +795,21 @@ class StandardSocketDeveloperCollector(DeveloperServicesCollector): def __init__( self, + destination_policy: DeveloperDestinationPolicy, allow_code_execution: bool = False, allow_state_change: bool = False, ) -> None: + if not isinstance(destination_policy, DeveloperDestinationPolicy): + raise DestinationScopeError("live collection requires a destination policy") + self.destination_policy = destination_policy self.allow_code_execution = allow_code_execution self.allow_state_change = allow_state_change def resolve_target(self, target_host: str) -> str: - return socket.gethostbyname(target_host) + self.destination_policy.authorize_host(target_host) + resolved_ip = socket.gethostbyname(target_host) + self.destination_policy.authorize_ip(resolved_ip) + return resolved_ip def probe_service( self, @@ -603,6 +823,8 @@ def probe_service( canary_artifact: str | None = None, timeout: float = 2.0, ) -> DeveloperServiceAssessment: + self.destination_policy.authorize_host(target_host) + self.destination_policy.authorize_ip(resolved_ip) try: with socket.create_connection((resolved_ip, port), timeout=timeout): return DeveloperServiceAssessment( @@ -623,7 +845,9 @@ def probe_service( can_change_state=False, bound_to_plan=True, configuration_details={"socket_connected": True}, - uncertainty_notes=["TCP connect succeeded; protocol-level handshake required to verify authentication"], + uncertainty_notes=[ + "TCP connect succeeded; protocol-level handshake required to verify authentication" + ], ) except (TimeoutError, ConnectionRefusedError, OSError) as err: return DeveloperServiceAssessment( @@ -659,12 +883,17 @@ def assess_developer_services( timeout: float = 2.0, allow_code_execution: bool = False, allow_state_change: bool = False, + destination_policy: DeveloperDestinationPolicy | None = None, ) -> DeveloperServicesReport: """Execute exposure and privilege boundary assessment across developer and runtime interfaces.""" active_collector = collector or StandardSocketDeveloperCollector( + destination_policy=destination_policy, allow_code_execution=allow_code_execution, allow_state_change=allow_state_change, ) + if isinstance(active_collector, StandardSocketDeveloperCollector): + for target in targets: + active_collector.destination_policy.authorize_host(target) svcs = service_types or list(DEFAULT_DEVELOPER_PORTS.keys()) canary = canary_artifact or canary_id diff --git a/cops/discovery/developer_destinations.py b/cops/discovery/developer_destinations.py new file mode 100644 index 00000000..e50c8685 --- /dev/null +++ b/cops/discovery/developer_destinations.py @@ -0,0 +1,200 @@ +"""Exact destination scope and offline routing for developer interface work.""" + +from __future__ import annotations + +import ipaddress +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + + +class DestinationScopeError(ValueError): + """A requested destination is outside the declared engagement scope.""" + + +_POLICY_KEYS = frozenset({"hosts", "resolved_ips", "repositories", "images", "runtimes", "clusters"}) +_HOST_LABEL = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$") + + +def _host(value: str) -> str: + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError("host must be a nonempty hostname or IP literal") + try: + return str(ipaddress.ip_address(value)) + except ValueError: + pass + host = value.lower() + if len(host) > 253 or any(not _HOST_LABEL.fullmatch(label) for label in host.split(".")): + raise ValueError("host must be an exact hostname or IP literal") + return host + + +def _exact(value: str, dimension: str) -> str: + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError(f"{dimension} must be a nonempty exact value") + if any(char in value for char in ("*", "?", "[", "]")) or any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError(f"{dimension} must not contain a pattern or control character") + return value + + +@dataclass(frozen=True) +class DeveloperDestinationPolicy: + """Explicit exact-match allowlists for all developer workflow destinations.""" + + hosts: frozenset[str] = frozenset() + resolved_ips: frozenset[str] = frozenset() + repositories: frozenset[str] = frozenset() + images: frozenset[str] = frozenset() + runtimes: frozenset[str] = frozenset() + clusters: frozenset[str] = frozenset() + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> DeveloperDestinationPolicy: + if not isinstance(data, dict) or set(data) - _POLICY_KEYS: + raise ValueError("destination policy contains unknown keys or is not an object") + values: dict[str, frozenset[str]] = {} + for key in _POLICY_KEYS: + entries = data.get(key, []) + if not isinstance(entries, list) or any(not isinstance(entry, str) for entry in entries): + raise ValueError(f"{key} must be a list of exact strings") + if key == "hosts": + values[key] = frozenset(_host(entry) for entry in entries) + elif key == "resolved_ips": + try: + values[key] = frozenset(str(ipaddress.ip_address(entry)) for entry in entries) + except ValueError as err: + raise ValueError("resolved_ips must contain exact IP addresses") from err + else: + values[key] = frozenset(_exact(entry, key) for entry in entries) + return cls(**values) + + @classmethod + def load(cls, path: str | Path) -> DeveloperDestinationPolicy: + return cls.from_dict(json.loads(Path(path).read_text(encoding="utf-8"))) + + def authorize_host(self, host: str) -> None: + if _host(host) not in self.hosts: + raise DestinationScopeError("host is outside the destination allowlist") + + def authorize_ip(self, address: str) -> None: + try: + canonical = str(ipaddress.ip_address(address)) + except ValueError as err: + raise DestinationScopeError("resolved address is not an IP literal") from err + if canonical not in self.resolved_ips: + raise DestinationScopeError("resolved address is outside the destination allowlist") + + def authorize(self, dimension: str, value: str) -> None: + if dimension not in {"repositories", "images", "runtimes", "clusters"}: + raise ValueError("unknown destination dimension") + if _exact(value, dimension) not in getattr(self, dimension): + raise DestinationScopeError(f"{dimension} destination is outside the allowlist") + + +def _named_entry(entries: Any, name: str, field: str) -> dict[str, Any]: + if not isinstance(entries, list): + raise DestinationScopeError(f"Kubernetes {field} list is missing") + matches = [item for item in entries if isinstance(item, dict) and item.get("name") == name] + if len(matches) != 1 or not isinstance(matches[0].get(field), dict): + raise DestinationScopeError(f"Kubernetes {field} entry is missing or ambiguous") + return matches[0][field] + + +def prepare_developer_handoff( + *, + kind: str, + target_host: str, + policy: DeveloperDestinationPolicy, + repository: str | None = None, + image: str | None = None, + runtime: str | None = None, + cluster: str | None = None, + kubeconfig: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Check destinations and Kubernetes metadata without contacting a service.""" + routes = { + "application": "cops-web-specialist", + "host": "cops-pentest-specialist", + "cluster": "cops-cloud-specialist", + } + if kind not in routes: + raise ValueError("handoff kind must be application, host, or cluster") + policy.authorize_host(target_host) + supplied = {"repositories": repository, "images": image, "runtimes": runtime, "clusters": cluster} + for dimension, value in supplied.items(): + if value is not None: + policy.authorize(dimension, value) + + handoff: dict[str, Any] = { + "specialist": routes[kind], + "kind": kind, + "target_host": _host(target_host), + "repository": repository, + "image": image, + "runtime": runtime, + "cluster": cluster, + "connectivity_verified": False, + } + if kind != "cluster": + if cluster is not None: + raise DestinationScopeError("cluster destination requires a cluster handoff") + if kubeconfig is not None: + raise DestinationScopeError("Kubernetes context only applies to cluster handoff") + return handoff + + if cluster is None or not isinstance(kubeconfig, dict): + raise DestinationScopeError("cluster handoff requires an allowed cluster and Kubernetes context") + context_name = kubeconfig.get("current-context") + if not isinstance(context_name, str) or not context_name: + raise DestinationScopeError("Kubernetes current-context is missing") + try: + _exact(context_name, "Kubernetes context") + except ValueError as err: + raise DestinationScopeError("Kubernetes current-context is invalid") from err + context = _named_entry(kubeconfig.get("contexts"), context_name, "context") + if context.get("cluster") != cluster: + raise DestinationScopeError("Kubernetes context does not select the allowed cluster") + cluster_data = _named_entry(kubeconfig.get("clusters"), cluster, "cluster") + server = cluster_data.get("server") + if not isinstance(server, str): + raise DestinationScopeError("Kubernetes API server is missing") + if any(ord(char) < 32 or ord(char) == 127 for char in server): + raise DestinationScopeError("Kubernetes API server is invalid") + try: + parsed = urlsplit(server) + except ValueError as err: + raise DestinationScopeError("Kubernetes API server is invalid") from err + try: + valid_port = parsed.port is None or 1 <= parsed.port <= 65535 + except ValueError as err: + raise DestinationScopeError("Kubernetes API server port is invalid") from err + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username + or parsed.password + or parsed.query + or parsed.fragment + or parsed.path not in ("", "/") + or not valid_port + ): + raise DestinationScopeError("Kubernetes API server must be an HTTPS origin") + policy.authorize_host(parsed.hostname) + namespace = context.get("namespace", "default") + if not isinstance(namespace, str) or not namespace: + raise DestinationScopeError("Kubernetes namespace is invalid") + try: + _exact(namespace, "Kubernetes namespace") + except ValueError as err: + raise DestinationScopeError("Kubernetes namespace is invalid") from err + handoff.update( + { + "kubernetes_context": context_name, + "kubernetes_namespace": namespace, + "kubernetes_api_host": _host(parsed.hostname), + } + ) + return handoff diff --git a/plugins/offensive-security/attack-surface-planner/README.md b/plugins/offensive-security/attack-surface-planner/README.md index 3ac28256..abd5fd88 100644 --- a/plugins/offensive-security/attack-surface-planner/README.md +++ b/plugins/offensive-security/attack-surface-planner/README.md @@ -295,12 +295,32 @@ The package evaluates exposure, access control, and code execution risk across d 6. **Developer Privilege Candidate Routing**: - Discovered misconfigurations (e.g. `docker_socket_rce`, `docker_registry_leak`, `rmi_code_execution`, `jdwp_code_execution`, `erlang_epmd_rce`, `adb_shell_rce`, `distcc_rce`, `svn_anonymous_checkout`, `ajp_ghostcat_rce`, `fastcgi_rce`) are structured as `DeveloperPrivilegeCandidate` records with cryptographic evidence hashes, auth prerequisites, and privilege impact ratings for handoff to offensive specialists (`cops-pentest-specialist`, `cops-redteam-operator`). +Live collection requires an exact destination policy. `hosts` are checked before DNS and `resolved_ips` after DNS, before any socket connection. Repository, image, runtime, and cluster entries are checked when supplied to the offline handoff. Omitted entries grant no permission for a later action. Keep this policy separate from engagement authorization and update it when approved destinations change. + +```json +{ + "hosts": ["dev01.corp.internal", "api.lab.internal"], + "resolved_ips": ["198.51.100.50"], + "repositories": ["team/service"], + "images": ["registry.internal/team/service:v1"], + "runtimes": ["containerd"], + "clusters": ["lab-cluster"] +} +``` + +The cluster handoff reads a JSON kubeconfig and requires its current context to select the allowed cluster and an HTTPS API server whose hostname is allowed. It does not connect to the cluster or validate credentials. Use a local kubeconfig file with appropriate access controls; the output contains only the selected context, namespace, and API hostname. + +The isolated CLI fixture in `tests/fixtures/developer_runtime_destinations.json` exercises an allowed handoff and negative controls for repository, image, runtime, cluster, Kubernetes context, and API hostname. Its test records rejection results in a temporary directory and verifies cleanup. This is offline process isolation, not a container or live-cluster check. + ```bash -# Assess developer and runtime interface services -python3 -m cops developer-services assess --targets "198.51.100.50,dev01.corp.internal" --vantage internal --canary-id "canary_docker_probe" --output developer_assessment.json +# Assess developer and runtime interface services with a live socket collector +python3 -m cops developer-services assess --mode live --targets "dev01.corp.internal" --destination-policy destination-policy.json --vantage internal --canary-id "canary_docker_probe" --output developer_assessment.json # Assess with authorized code execution verification -python3 -m cops developer-services assess --targets "198.51.100.50" --services "docker,jdwp,distcc" --allow-code-execution --output developer_assessment.json +python3 -m cops developer-services assess --mode live --targets "dev01.corp.internal" --destination-policy destination-policy.json --services "docker,jdwp,distcc" --allow-code-execution --output developer_assessment.json + +# Prepare a scope-checked cluster handoff using a JSON kubeconfig +python3 -m cops developer-services handoff --kind cluster --target-host dev01.corp.internal --destination-policy destination-policy.json --repository team/service --image registry.internal/team/service:v1 --runtime containerd --cluster lab-cluster --kubeconfig kubeconfig.json --output cluster_handoff.json # Extract developer privilege candidates for code execution or breakout workflows python3 -m cops developer-services candidates developer_assessment.json --output developer_candidates.json diff --git a/plugins/offensive-security/attack-surface-planner/docs/PLAYBOOK.md b/plugins/offensive-security/attack-surface-planner/docs/PLAYBOOK.md index 59f59ed1..b9d3055b 100644 --- a/plugins/offensive-security/attack-surface-planner/docs/PLAYBOOK.md +++ b/plugins/offensive-security/attack-surface-planner/docs/PLAYBOOK.md @@ -151,7 +151,7 @@ Invoke this planner under the following concrete triggers: - Declare `ExecutionEffect`: `read_only`, `non_destructive`, `state_change`, `code_execution`. - Operations that execute code or modify state require explicit authorization (`--allow-code-execution`, `--allow-state-change`) bound to the active plan. 3. **Application vs Infrastructure Specialist Routing**: - - Route application-layer findings to web specialists (`cops-web-specialist`), orchestrator paths to cloud specialists (`cops-cloud-specialist`), and code execution / breakout findings to pentest specialists (`cops-pentest-specialist`). + - Route application-layer findings to web specialists (`cops-web-specialist`), orchestrator paths to cloud specialists (`cops-cloud-specialist`), and code execution / breakout findings to pentest specialists (`cops-pentest-specialist`). Before a cluster handoff, validate the allowed cluster, current Kubernetes context, and API hostname with `python3 -m cops developer-services handoff --kind cluster ...`; this is an offline metadata check. 4. **Strict Inaccessible != Secure Grounding**: - Filtered, closed, or timed-out endpoints are strictly classified as `inaccessible` (with `auth_prerequisite: unknown`). Never assume an inaccessible service is secure or hardened. 5. **Canary Validation and Verifiable Cleanup Receipts**: diff --git a/plugins/offensive-security/attack-surface-planner/skills/network-developer-interfaces/SKILL.md b/plugins/offensive-security/attack-surface-planner/skills/network-developer-interfaces/SKILL.md index 9eca3363..b6fa5e6c 100644 --- a/plugins/offensive-security/attack-surface-planner/skills/network-developer-interfaces/SKILL.md +++ b/plugins/offensive-security/attack-surface-planner/skills/network-developer-interfaces/SKILL.md @@ -15,6 +15,7 @@ description: Assess developer and runtime interfaces (Docker, Docker Registry, R 3. **Application vs Infrastructure Specialist Routing**: - Route application behavior to web/API workflows and cluster paths to Kubernetes context establishment (`cops-web-specialist`, `cops-cloud-specialist`, `cops-pentest-specialist`). + - Prepare cluster handoff with an exact destination policy and a JSON kubeconfig. The handoff validates the current context and allowed API hostname offline; it does not verify connectivity. 4. **Inaccessible != Secure Truth Boundary**: - Strictly mark unreachable, filtered, or connection-refused services as `inaccessible` (with `auth_prerequisite: unknown`). Never claim unverified services are protected or secure. @@ -25,7 +26,8 @@ description: Assess developer and runtime interfaces (Docker, Docker Registry, R 6. **CLI Invocation**: ```bash - python3 -m cops developer-services assess --targets dev01.corp.internal --canary-id "canary_docker_probe" --output dev_report.json + python3 -m cops developer-services assess --mode live --targets dev01.corp.internal --destination-policy destination-policy.json --canary-id "canary_docker_probe" --output dev_report.json + python3 -m cops developer-services handoff --kind cluster --target-host dev01.corp.internal --destination-policy destination-policy.json --cluster lab-cluster --kubeconfig kubeconfig.json --output cluster_handoff.json python3 -m cops developer-services candidates dev_report.json --output dev_candidates.json python3 -m cops developer-services cleanup dev_report.json --output cleanup_receipts.json python3 -m cops developer-services inspect dev_report.json diff --git a/specs/features/network_developer_services.feature b/specs/features/network_developer_services.feature index 53f5e959..34530ae0 100644 --- a/specs/features/network_developer_services.feature +++ b/specs/features/network_developer_services.feature @@ -42,3 +42,19 @@ Feature: Developer and Runtime Interfaces Assessment Then the exposure status is reported as "protected" And authentication prerequisites reflect "client_cert", "token_or_api_key", or "user_password" And zero unauthenticated developer privilege candidates are generated + + Scenario: Blocking unlisted live destinations before network access + Given an exact developer destination policy for a lab host and resolved address + When the live collector receives an unlisted host and an unlisted resolved address + Then the host is rejected before DNS and the address before a socket opens + + Scenario: Checking Kubernetes context and all runtime destinations before cluster handoff + Given an exact developer destination policy and Kubernetes context for a lab cluster + When the approved cluster handoff and out-of-scope variants are prepared offline + Then the approved handoff routes to the cloud specialist without claiming connectivity + And unlisted repositories, images, runtimes, clusters, contexts, and API hosts are rejected + + Scenario: Recording a rejected offline runtime handoff and cleanup + Given an isolated offline developer runtime fixture + When the fixture runs an approved handoff and an unlisted repository control + Then it records the rejection and verifies temporary fixture cleanup diff --git a/tests/fixtures/developer_runtime_destinations.json b/tests/fixtures/developer_runtime_destinations.json new file mode 100644 index 00000000..ffb230e5 --- /dev/null +++ b/tests/fixtures/developer_runtime_destinations.json @@ -0,0 +1,31 @@ +{ + "policy": { + "hosts": ["dev.corp.internal", "api.cluster.internal"], + "resolved_ips": ["198.51.100.50"], + "repositories": ["team/service"], + "images": ["registry.internal/team/service:v1"], + "runtimes": ["containerd"], + "clusters": ["lab-cluster"] + }, + "kubeconfig": { + "current-context": "lab-context", + "contexts": [ + {"name": "lab-context", "context": {"cluster": "lab-cluster", "namespace": "testing"}} + ], + "clusters": [ + {"name": "lab-cluster", "cluster": {"server": "https://api.cluster.internal:6443"}} + ] + }, + "allowed_destinations": { + "repository": "team/service", + "image": "registry.internal/team/service:v1", + "runtime": "containerd", + "cluster": "lab-cluster" + }, + "negative_controls": [ + {"id": "repository", "option": "--repository", "value": "other/repo", "error": "repositories destination is outside the allowlist"}, + {"id": "image", "option": "--image", "value": "registry.internal/other:v1", "error": "images destination is outside the allowlist"}, + {"id": "runtime", "option": "--runtime", "value": "docker", "error": "runtimes destination is outside the allowlist"}, + {"id": "cluster", "option": "--cluster", "value": "production", "error": "clusters destination is outside the allowlist"} + ] +} diff --git a/tests/step_defs/test_network_developer_services_steps.py b/tests/step_defs/test_network_developer_services_steps.py index 8d287c73..3a24b765 100644 --- a/tests/step_defs/test_network_developer_services_steps.py +++ b/tests/step_defs/test_network_developer_services_steps.py @@ -4,8 +4,14 @@ from __future__ import annotations +import copy +import json +import os +import subprocess import sys +import tempfile from pathlib import Path +from unittest.mock import patch import pytest from pytest_bdd import given, scenarios, then, when @@ -21,8 +27,14 @@ DeveloperPrivilegeImpact, ExecutionEffect, OfflineSyntheticDeveloperCollector, + StandardSocketDeveloperCollector, assess_developer_services, ) +from cops.discovery.developer_destinations import ( + DestinationScopeError, + DeveloperDestinationPolicy, + prepare_developer_handoff, +) scenarios("../../specs/features/network_developer_services.feature") @@ -33,7 +45,12 @@ def bdd_ctx(): "targets": ["vulnerable-dev.corp.internal"], "targets_db": { "vulnerable-dev.corp.internal": { - "docker": {"socket_exposed": True, "tls_verify": False, "auth_required": False, "version": "Docker 20.10"}, + "docker": { + "socket_exposed": True, + "tls_verify": False, + "auth_required": False, + "version": "Docker 20.10", + }, "docker_registry": {"catalog_accessible": True, "auth_required": False, "version": "Registry 2.8"}, "rmi": {"unauthenticated_registry": True, "codebase_only": False, "version": "Java RMI 11"}, "jdwp": {"debug_agent_exposed": True, "auth_required": False, "version": "JDWP 1.6"}, @@ -45,10 +62,33 @@ def bdd_ctx(): "fastcgi": {"exposed_external": True, "auth_required": False, "version": "PHP-FPM 7.4"}, }, "hardened-dev.corp.internal": { - "docker": {"protected": True, "tls_verify": True, "auth_required": True, "auth_prerequisite": "client_cert", "version": "Docker 24.0"}, - "docker_registry": {"protected": True, "auth_required": True, "auth_prerequisite": "token_or_api_key", "version": "Registry 2.8.2"}, - "jdwp": {"protected": True, "debug_agent_exposed": False, "auth_required": True, "auth_prerequisite": "user_password", "version": "JDWP Disabled"}, - "ajp": {"protected": True, "secret_required": True, "auth_required": True, "auth_prerequisite": "user_password", "version": "Tomcat 9.0.31"}, + "docker": { + "protected": True, + "tls_verify": True, + "auth_required": True, + "auth_prerequisite": "client_cert", + "version": "Docker 24.0", + }, + "docker_registry": { + "protected": True, + "auth_required": True, + "auth_prerequisite": "token_or_api_key", + "version": "Registry 2.8.2", + }, + "jdwp": { + "protected": True, + "debug_agent_exposed": False, + "auth_required": True, + "auth_prerequisite": "user_password", + "version": "JDWP Disabled", + }, + "ajp": { + "protected": True, + "secret_required": True, + "auth_required": True, + "auth_prerequisite": "user_password", + "version": "Tomcat 9.0.31", + }, }, "filtered-dev.corp.internal": { "docker": {"state": "filtered"}, @@ -64,7 +104,9 @@ def bdd_ctx(): # Scenario 1: Assessing developer and runtime interface services -@given("an approved target host exposing Docker, Docker Registry, RMI, JDWP, Erlang EPMD, ADB, distcc, SVN, AJP, and FastCGI") +@given( + "an approved target host exposing Docker, Docker Registry, RMI, JDWP, Erlang EPMD, ADB, distcc, SVN, AJP, and FastCGI" +) def setup_multi_protocol_target(bdd_ctx): bdd_ctx["collector"] = OfflineSyntheticDeveloperCollector( targets=bdd_ctx["targets_db"], @@ -84,7 +126,9 @@ def execute_multi_protocol_probes(bdd_ctx): ) -@then("discrete assessments are recorded across container orchestration, language debug, distributed build SCM, and application gateway categories") +@then( + "discrete assessments are recorded across container orchestration, language debug, distributed build SCM, and application gateway categories" +) def verify_discrete_categories(bdd_ctx): report = bdd_ctx["report"] categories = {a.category for a in report.assessments} @@ -231,7 +275,9 @@ def extract_candidates(bdd_ctx): bdd_ctx["candidates"] = candidates -@then('actionable candidates for "docker_socket_rce", "jdwp_code_execution", "distcc_rce", and "fastcgi_rce" are generated') +@then( + 'actionable candidates for "docker_socket_rce", "jdwp_code_execution", "distcc_rce", and "fastcgi_rce" are generated' +) def verify_expected_candidates(bdd_ctx): types = [c.finding_type for c in bdd_ctx["candidates"]] assert "docker_socket_rce" in types @@ -240,7 +286,9 @@ def verify_expected_candidates(bdd_ctx): assert "fastcgi_rce" in types -@then("each candidate contains service type, target host, port, execution effect, privilege impact, SHA-256 evidence hash, and remediation guidance") +@then( + "each candidate contains service type, target host, port, execution effect, privilege impact, SHA-256 evidence hash, and remediation guidance" +) def verify_candidate_integrity(bdd_ctx): for c in bdd_ctx["candidates"]: assert c.service_type @@ -291,3 +339,175 @@ def verify_auth_prerequisites(bdd_ctx): @then("zero unauthenticated developer privilege candidates are generated") def verify_zero_candidates(bdd_ctx): assert bdd_ctx["report"].candidates_count == 0 + + +@given("an exact developer destination policy for a lab host and resolved address") +def setup_live_destination_policy(bdd_ctx): + bdd_ctx["destination_policy"] = DeveloperDestinationPolicy.from_dict( + { + "hosts": ["dev.corp.internal"], + "resolved_ips": ["198.51.100.50"], + } + ) + + +@when("the live collector receives an unlisted host and an unlisted resolved address") +def try_unlisted_live_destinations(bdd_ctx): + collector = StandardSocketDeveloperCollector(destination_policy=bdd_ctx["destination_policy"]) + with patch("socket.gethostbyname") as dns, patch("socket.create_connection") as connect: + with pytest.raises(DestinationScopeError): + collector.resolve_target("outside.corp.internal") + bdd_ctx["host_blocked_before_dns"] = dns.call_count == 0 + dns.return_value = "203.0.113.99" + with pytest.raises(DestinationScopeError): + collector.assess_docker("dev.corp.internal") + bdd_ctx["address_blocked_before_socket"] = not connect.called + + +@then("the host is rejected before DNS and the address before a socket opens") +def verify_live_destination_gate(bdd_ctx): + assert bdd_ctx["host_blocked_before_dns"] + assert bdd_ctx["address_blocked_before_socket"] + + +@given("an exact developer destination policy and Kubernetes context for a lab cluster") +def setup_cluster_destination_policy(bdd_ctx): + fixture = json.loads((ROOT / "tests/fixtures/developer_runtime_destinations.json").read_text(encoding="utf-8")) + bdd_ctx["runtime_fixture"] = fixture + bdd_ctx["destination_policy"] = DeveloperDestinationPolicy.from_dict(fixture["policy"]) + + +@when("the approved cluster handoff and out-of-scope variants are prepared offline") +def prepare_cluster_variants(bdd_ctx): + fixture = bdd_ctx["runtime_fixture"] + base = { + "kind": "cluster", + "target_host": "dev.corp.internal", + "policy": bdd_ctx["destination_policy"], + "kubeconfig": fixture["kubeconfig"], + **fixture["allowed_destinations"], + } + bdd_ctx["approved_handoff"] = prepare_developer_handoff(**base) + rejected = set() + for control in fixture["negative_controls"]: + changed = dict(base) + changed[control["option"].removeprefix("--")] = control["value"] + with pytest.raises(DestinationScopeError, match=control["error"]): + prepare_developer_handoff(**changed) + rejected.add(control["id"]) + + wrong_context = copy.deepcopy(fixture["kubeconfig"]) + wrong_context["current-context"] = "wrong-context" + with pytest.raises(DestinationScopeError, match="Kubernetes context entry is missing"): + prepare_developer_handoff(**{**base, "kubeconfig": wrong_context}) + rejected.add("kubernetes-context") + + wrong_api = copy.deepcopy(fixture["kubeconfig"]) + wrong_api["clusters"][0]["cluster"]["server"] = "https://outside.cluster.internal:6443" + with pytest.raises(DestinationScopeError, match="host is outside the destination allowlist"): + prepare_developer_handoff(**{**base, "kubeconfig": wrong_api}) + rejected.add("kubernetes-api-host") + bdd_ctx["rejected_variants"] = rejected + + +@then("the approved handoff routes to the cloud specialist without claiming connectivity") +def verify_approved_cluster_handoff(bdd_ctx): + handoff = bdd_ctx["approved_handoff"] + assert handoff["specialist"] == "cops-cloud-specialist" + assert handoff["kubernetes_context"] == "lab-context" + assert handoff["kubernetes_api_host"] == "api.cluster.internal" + assert handoff["connectivity_verified"] is False + + +@then("unlisted repositories, images, runtimes, clusters, contexts, and API hosts are rejected") +def verify_rejected_cluster_variants(bdd_ctx): + assert bdd_ctx["rejected_variants"] == { + "repository", + "image", + "runtime", + "cluster", + "kubernetes-context", + "kubernetes-api-host", + } + + +@given("an isolated offline developer runtime fixture") +def setup_offline_runtime_fixture(bdd_ctx): + bdd_ctx["runtime_fixture"] = json.loads( + (ROOT / "tests/fixtures/developer_runtime_destinations.json").read_text(encoding="utf-8") + ) + + +@when("the fixture runs an approved handoff and an unlisted repository control") +def run_offline_runtime_fixture(bdd_ctx): + fixture = bdd_ctx["runtime_fixture"] + with tempfile.TemporaryDirectory() as record_dir: + with tempfile.TemporaryDirectory(dir=record_dir) as work_dir: + work_path = Path(work_dir) + policy_file = work_path / "scope.json" + config_file = work_path / "kubeconfig.json" + output_file = work_path / "handoff.json" + policy_file.write_text(json.dumps(fixture["policy"]), encoding="utf-8") + config_file.write_text(json.dumps(fixture["kubeconfig"]), encoding="utf-8") + command = [ + sys.executable, + "-m", + "cops", + "developer-services", + "handoff", + "--kind", + "cluster", + "--target-host", + "dev.corp.internal", + "--destination-policy", + str(policy_file), + "--kubeconfig", + str(config_file), + "--output", + str(output_file), + ] + allowed = [ + item for option, value in fixture["allowed_destinations"].items() for item in (f"--{option}", value) + ] + child_env = {key: os.environ[key] for key in ("PATH", "SYSTEMROOT", "WINDIR") if key in os.environ} + child_env.update({"PYTHONPATH": str(ROOT), "HOME": work_dir, "TMPDIR": work_dir}) + + def run(arguments): + return subprocess.run( + [*command, *arguments], + cwd=work_path, + env=child_env, + capture_output=True, + text=True, + check=False, + timeout=15, + ) + + accepted = run(allowed) + assert accepted.returncode == 0, accepted.stderr + assert output_file.is_file() + output_file.unlink() + control = fixture["negative_controls"][0] + blocked_args = [ + item + for option, value in fixture["allowed_destinations"].items() + if option != "repository" + for item in (f"--{option}", value) + ] + blocked = run([*blocked_args, control["option"], control["value"]]) + bdd_ctx["runtime_record"] = { + "exit_code": blocked.returncode, + "reason": control["error"] if control["error"] in blocked.stderr else None, + "output_created": output_file.exists(), + } + bdd_ctx["runtime_record"]["cleanup_verified"] = not work_path.exists() + + +@then("it records the rejection and verifies temporary fixture cleanup") +def verify_offline_runtime_fixture(bdd_ctx): + assert bdd_ctx["runtime_record"] == { + "exit_code": 2, + "reason": "repositories destination is outside the allowlist", + "output_created": False, + "cleanup_verified": True, + } diff --git a/tests/test_developer_services.py b/tests/test_developer_services.py index c62749ce..361d2b2a 100644 --- a/tests/test_developer_services.py +++ b/tests/test_developer_services.py @@ -5,7 +5,11 @@ from __future__ import annotations import argparse +import contextlib +import io import json +import os +import subprocess import sys import tempfile import unittest @@ -30,8 +34,14 @@ assess_developer_services, ) from cops.discovery.cli import ( + build_developer_parser, command_developer_discovery, ) +from cops.discovery.developer_destinations import ( + DestinationScopeError, + DeveloperDestinationPolicy, + prepare_developer_handoff, +) class TestDeveloperModelsAndSerialization(unittest.TestCase): @@ -335,7 +345,9 @@ def test_code_execution_gating_unauthorized(self): res = strict_collector.assess_docker("vulnerable-dev.corp.internal", canary_artifact="canary_docker") self.assertFalse(res.can_execute_code) self.assertFalse(res.can_change_state) - self.assertIn("Code execution verification skipped: not authorized in active action plan", res.uncertainty_notes) + self.assertIn( + "Code execution verification skipped: not authorized in active action plan", res.uncertainty_notes + ) self.assertIn("State change verification skipped: not authorized in active action plan", res.uncertainty_notes) def test_protected_service_assessment(self): @@ -381,7 +393,10 @@ def test_inaccessible_filtered_service(self): self.assertEqual(res.exposure_status, DeveloperExposureStatus.INACCESSIBLE.value) self.assertEqual(res.auth_prerequisite, DeveloperAuthPrerequisite.UNKNOWN.value) self.assertFalse(res.canary_validated) - self.assertIn("Service was inaccessible from probe vantage; this cannot be reported as secure or hardened", res.uncertainty_notes[0]) + self.assertIn( + "Service was inaccessible from probe vantage; this cannot be reported as secure or hardened", + res.uncertainty_notes[0], + ) self.assertEqual(len(res.privilege_candidates), 0) def test_inaccessible_timeout_service(self): @@ -520,6 +535,9 @@ def test_socket_probe_connected(self, mock_gethost, mock_conn): mock_conn.return_value.__enter__.return_value = MagicMock() collector = StandardSocketDeveloperCollector( + destination_policy=DeveloperDestinationPolicy.from_dict( + {"hosts": ["dev.corp.internal"], "resolved_ips": ["198.51.100.50"]} + ), allow_code_execution=True, allow_state_change=True, ) @@ -535,12 +553,383 @@ def test_socket_probe_connection_refused(self, mock_gethost, mock_conn): mock_gethost.return_value = "198.51.100.50" mock_conn.side_effect = ConnectionRefusedError("Connection refused") - collector = StandardSocketDeveloperCollector() + collector = StandardSocketDeveloperCollector( + destination_policy=DeveloperDestinationPolicy.from_dict( + {"hosts": ["dev.corp.internal"], "resolved_ips": ["198.51.100.50"]} + ), + ) res = collector.assess_docker("dev.corp.internal") self.assertEqual(res.exposure_status, DeveloperExposureStatus.INACCESSIBLE.value) self.assertEqual(res.auth_prerequisite, DeveloperAuthPrerequisite.UNKNOWN.value) self.assertIn("Connection refused", res.error_message) +class TestDeveloperDestinationPolicy(unittest.TestCase): + def setUp(self): + self.policy = DeveloperDestinationPolicy.from_dict( + { + "hosts": ["dev.corp.internal", "api.cluster.internal"], + "resolved_ips": ["198.51.100.50"], + "repositories": ["team/service"], + "images": ["registry.internal/team/service:v1"], + "runtimes": ["containerd"], + "clusters": ["lab-cluster"], + } + ) + + def test_policy_rejects_unknown_keys_and_wildcards(self): + with self.assertRaises(ValueError): + DeveloperDestinationPolicy.from_dict({"host": ["dev.corp.internal"]}) + with self.assertRaises(ValueError): + DeveloperDestinationPolicy.from_dict({"hosts": ["*.corp.internal"]}) + with self.assertRaises(ValueError): + DeveloperDestinationPolicy.from_dict({"resolved_ips": ["198.51.100.0/24"]}) + + @patch("socket.create_connection") + @patch("socket.gethostbyname") + def test_live_rejects_host_before_dns_and_ip_before_socket(self, mock_gethost, mock_conn): + collector = StandardSocketDeveloperCollector(destination_policy=self.policy) + with self.assertRaises(DestinationScopeError): + collector.assess_docker("outside.corp.internal") + mock_gethost.assert_not_called() + mock_conn.assert_not_called() + + mock_gethost.return_value = "203.0.113.99" + with self.assertRaises(DestinationScopeError): + collector.assess_docker("dev.corp.internal") + mock_conn.assert_not_called() + + with self.assertRaises(DestinationScopeError): + collector.probe_service("dev.corp.internal", "203.0.113.99", "docker", 2375) + mock_conn.assert_not_called() + + @patch("socket.gethostbyname") + def test_batch_rejects_out_of_scope_target_before_any_dns(self, mock_gethost): + collector = StandardSocketDeveloperCollector(destination_policy=self.policy) + with self.assertRaises(DestinationScopeError): + assess_developer_services(["dev.corp.internal", "outside.corp.internal"], collector=collector) + mock_gethost.assert_not_called() + + def test_cluster_handoff_requires_matching_kubernetes_context(self): + config = { + "current-context": "lab-context", + "contexts": [{"name": "lab-context", "context": {"cluster": "lab-cluster", "namespace": "testing"}}], + "clusters": [{"name": "lab-cluster", "cluster": {"server": "https://api.cluster.internal:6443"}}], + } + handoff = prepare_developer_handoff( + kind="cluster", + target_host="dev.corp.internal", + policy=self.policy, + repository="team/service", + image="registry.internal/team/service:v1", + runtime="containerd", + cluster="lab-cluster", + kubeconfig=config, + ) + self.assertEqual(handoff["specialist"], "cops-cloud-specialist") + self.assertEqual(handoff["kubernetes_context"], "lab-context") + self.assertFalse(handoff["connectivity_verified"]) + + config["contexts"][0]["context"]["cluster"] = "other-cluster" + with self.assertRaises(DestinationScopeError): + prepare_developer_handoff( + kind="cluster", + target_host="dev.corp.internal", + policy=self.policy, + runtime="containerd", + cluster="lab-cluster", + kubeconfig=config, + ) + + def test_handoff_rejects_unapproved_destinations_and_preserves_application_route(self): + for field, bad_value in ( + ("repository", "other/repo"), + ("image", "registry.internal/other:v1"), + ("runtime", "docker"), + ("cluster", "production"), + ): + kwargs = {field: bad_value} + with self.subTest(field=field), self.assertRaises(DestinationScopeError): + prepare_developer_handoff( + kind="application", + target_host="dev.corp.internal", + policy=self.policy, + **kwargs, + ) + handoff = prepare_developer_handoff( + kind="application", + target_host="dev.corp.internal", + policy=self.policy, + ) + self.assertEqual(handoff["specialist"], "cops-web-specialist") + with self.assertRaises(DestinationScopeError): + prepare_developer_handoff( + kind="application", + target_host="dev.corp.internal", + policy=self.policy, + cluster="lab-cluster", + ) + + def test_policy_rejects_control_characters_and_bad_kubernetes_server(self): + with self.assertRaises(ValueError): + DeveloperDestinationPolicy.from_dict({"images": ["registry.internal/team\tservice:v1"]}) + config = { + "current-context": "lab-context", + "contexts": [{"name": "lab-context", "context": {"cluster": "lab-cluster"}}], + "clusters": [{"name": "lab-cluster", "cluster": {"server": "https://[bad-host"}}], + } + with self.assertRaises(DestinationScopeError): + prepare_developer_handoff( + kind="cluster", + target_host="dev.corp.internal", + policy=self.policy, + cluster="lab-cluster", + kubeconfig=config, + ) + + config["clusters"][0]["cluster"]["server"] = "https://api.cluster.internal\n:6443" + with self.assertRaises(DestinationScopeError): + prepare_developer_handoff( + kind="cluster", + target_host="dev.corp.internal", + policy=self.policy, + cluster="lab-cluster", + kubeconfig=config, + ) + + config["clusters"][0]["cluster"]["server"] = "https://outside.cluster.internal:6443" + with self.assertRaisesRegex(DestinationScopeError, "host is outside the destination allowlist"): + prepare_developer_handoff( + kind="cluster", + target_host="dev.corp.internal", + policy=self.policy, + cluster="lab-cluster", + kubeconfig=config, + ) + + +class TestDeveloperDestinationCli(unittest.TestCase): + def setUp(self): + self.parser = argparse.ArgumentParser() + build_developer_parser(self.parser) + + @patch("socket.gethostbyname") + def test_live_requires_policy_and_rejects_unapproved_host_before_dns(self, mock_gethost): + with tempfile.TemporaryDirectory() as tmpdir: + policy_file = Path(tmpdir) / "scope.json" + policy_file.write_text( + json.dumps( + { + "hosts": ["dev.corp.internal"], + "resolved_ips": ["198.51.100.50"], + } + ), + encoding="utf-8", + ) + missing = self.parser.parse_args( + [ + "assess", + "--mode", + "live", + "--targets", + "outside.corp.internal", + "--services", + "docker", + ] + ) + self.assertEqual(command_developer_discovery(missing), 2) + blocked = self.parser.parse_args( + [ + "assess", + "--mode", + "live", + "--targets", + "outside.corp.internal", + "--services", + "docker", + "--destination-policy", + str(policy_file), + ] + ) + self.assertEqual(command_developer_discovery(blocked), 2) + mock_gethost.assert_not_called() + + def test_handoff_output_error_is_sanitized(self): + with tempfile.TemporaryDirectory() as tmpdir: + policy_file = Path(tmpdir) / "scope.json" + policy_file.write_text(json.dumps({"hosts": ["dev.corp.internal"]}), encoding="utf-8") + args = self.parser.parse_args( + [ + "handoff", + "--kind", + "application", + "--target-host", + "dev.corp.internal", + "--destination-policy", + str(policy_file), + "--output", + tmpdir, + ] + ) + error = io.StringIO() + with contextlib.redirect_stderr(error): + self.assertEqual(command_developer_discovery(args), 2) + self.assertEqual(error.getvalue().strip(), "Destination output error: unable to write handoff record") + self.assertNotIn(tmpdir, error.getvalue()) + + def test_live_policy_read_error_is_sanitized(self): + with tempfile.TemporaryDirectory() as tmpdir: + missing_policy = Path(tmpdir) / "missing-policy.json" + args = self.parser.parse_args( + [ + "assess", + "--mode", + "live", + "--targets", + "dev.corp.internal", + "--services", + "docker", + "--destination-policy", + str(missing_policy), + ] + ) + error = io.StringIO() + with contextlib.redirect_stderr(error): + self.assertEqual(command_developer_discovery(args), 2) + self.assertEqual(error.getvalue().strip(), "Destination scope error: unable to read destination policy") + self.assertNotIn(tmpdir, error.getvalue()) + + def test_isolated_runtime_destination_fixture_records_rejections_and_cleans_up(self): + fixture = json.loads((ROOT / "tests/fixtures/developer_runtime_destinations.json").read_text(encoding="utf-8")) + with tempfile.TemporaryDirectory() as record_dir: + evidence_file = Path(record_dir) / "runtime-evidence.json" + with tempfile.TemporaryDirectory(dir=record_dir) as tmpdir: + tmp_path = Path(tmpdir) + policy_file = tmp_path / "scope.json" + config_file = tmp_path / "kubeconfig.json" + output_file = tmp_path / "handoff.json" + policy_file.write_text(json.dumps(fixture["policy"]), encoding="utf-8") + config_file.write_text(json.dumps(fixture["kubeconfig"]), encoding="utf-8") + + base_command = [ + sys.executable, + "-m", + "cops", + "developer-services", + "handoff", + "--kind", + "cluster", + "--target-host", + "dev.corp.internal", + "--destination-policy", + str(policy_file), + "--kubeconfig", + str(config_file), + "--output", + str(output_file), + ] + allowed_args = [ + item for option, value in fixture["allowed_destinations"].items() for item in (f"--{option}", value) + ] + + def run_handoff(arguments): + child_env = {key: os.environ[key] for key in ("PATH", "SYSTEMROOT", "WINDIR") if key in os.environ} + child_env.update( + { + "PYTHONPATH": str(ROOT), + "HOME": str(tmp_path), + "XDG_CONFIG_HOME": str(tmp_path), + "TMPDIR": str(tmp_path), + } + ) + return subprocess.run( + [*base_command, *arguments], + cwd=tmp_path, + env=child_env, + capture_output=True, + text=True, + check=False, + timeout=15, + ) + + accepted = run_handoff(allowed_args) + self.assertEqual(accepted.returncode, 0, accepted.stderr) + result = json.loads(output_file.read_text(encoding="utf-8")) + self.assertEqual(result["specialist"], "cops-cloud-specialist") + self.assertEqual(result["kubernetes_context"], "lab-context") + self.assertFalse(result["connectivity_verified"]) + output_file.unlink() + + rejection_records = [] + for control in fixture["negative_controls"]: + blocked_args = [ + item + for option, value in fixture["allowed_destinations"].items() + if f"--{option}" != control["option"] + for item in (f"--{option}", value) + ] + blocked = run_handoff([*blocked_args, control["option"], control["value"]]) + self.assertEqual(blocked.returncode, 2, control["id"]) + self.assertIn(control["error"], blocked.stderr) + self.assertFalse(output_file.exists(), control["id"]) + rejection_records.append( + { + "id": control["id"], + "exit_code": blocked.returncode, + "reason": control["error"], + "output_created": False, + } + ) + + bad_config = dict(fixture["kubeconfig"]) + bad_config["current-context"] = "wrong-context" + config_file.write_text(json.dumps(bad_config), encoding="utf-8") + blocked_context = run_handoff(allowed_args) + self.assertEqual(blocked_context.returncode, 2) + context_error = "Kubernetes context entry is missing" + self.assertIn(context_error, blocked_context.stderr) + self.assertFalse(output_file.exists()) + rejection_records.append( + { + "id": "kubernetes-context", + "exit_code": blocked_context.returncode, + "reason": context_error, + "output_created": False, + } + ) + + bad_config["current-context"] = fixture["kubeconfig"]["current-context"] + bad_config["clusters"][0]["cluster"]["server"] = "https://outside.cluster.internal:6443" + config_file.write_text(json.dumps(bad_config), encoding="utf-8") + blocked_api = run_handoff(allowed_args) + self.assertEqual(blocked_api.returncode, 2) + api_error = "host is outside the destination allowlist" + self.assertIn(api_error, blocked_api.stderr) + self.assertFalse(output_file.exists()) + rejection_records.append( + { + "id": "kubernetes-api-host", + "exit_code": blocked_api.returncode, + "reason": api_error, + "output_created": False, + } + ) + + cleanup_verified = not tmp_path.exists() + self.assertTrue(cleanup_verified) + evidence = { + "fixture": "developer_runtime_destinations.json", + "allowed_handoff": { + "exit_code": accepted.returncode, + "connectivity_verified": result["connectivity_verified"], + }, + "negative_controls": rejection_records, + "cleanup_verified": cleanup_verified, + } + evidence_file.write_text(json.dumps(evidence, indent=2), encoding="utf-8") + self.assertEqual(json.loads(evidence_file.read_text(encoding="utf-8")), evidence) + self.assertEqual(len(rejection_records), 6) + + if __name__ == "__main__": unittest.main()