feature: add SSRF destination validation detection rule (CWE-918)#222
feature: add SSRF destination validation detection rule (CWE-918)#222ticket-fixer[bot] wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
Code Review Summary
Total Issues Found: 5
Critical Issues: 2
Suggestions: 3
Key Findings:
The code demonstrates solid SSRF security awareness but has two critical issues: (1) a TOCTOU vulnerability and performance problem from duplicate DNS resolution between accept() and check(), and (2) incomplete exception handling that only catches socket.gaierror instead of OSError, risking unhandled crashes. Additionally, there are suggestions to fix a type hint inconsistency (str vs None check), remove a redundant test that violates conventions by testing a private method, and document or remove the redundant safety check in check(). The most impactful fix is caching the DNS resolution result across accept/check to resolve both the TOCTOU and performance concerns simultaneously.
| privacy_issue=False, | ||
| has_public_exploit=False, | ||
| targeted_by_malware=False, | ||
| targeted_by_ransomware=False, |
There was a problem hiding this comment.
[Performance / Critical] Cache the result of _is_unsafe_destination() in accept() as an instance attribute (self._cached_is_unsafe), then reference that cached value in check() instead of calling _is_unsafe_destination() again. This eliminates the duplicate DNS resolution (the most expensive operation in this rule) and also prevents TOCTOU issues where DNS could return different results between the two calls. Initialize self._cached_is_unsafe = False in __init__ for safety. In the check method (~line 197), replace the call to _is_unsafe_destination(target.host) with self._cached_is_unsafe.
| targeted_by_ransomware=False, | |
| def accept(self, target: definitions.Target) -> bool: | |
| """Accept targets that resolve to an unsafe destination. | |
| Args: | |
| target: The target to verify. | |
| Returns: | |
| True if the target host resolves to a private, loopback, | |
| link-local, reserved or cloud-metadata destination. | |
| """ | |
| self._cached_is_unsafe = _is_unsafe_destination(target.host) | |
| return self._cached_is_unsafe |
| pass | ||
|
|
||
| try: | ||
| infos = socket.getaddrinfo(host, None) |
There was a problem hiding this comment.
[Bug] The function signature declares host: str, but the body checks if host is None: on line 130. This is a type contract inconsistency. If None is a possible input (defensive programming), the type hint should be host: str | None. If None is truly not possible (since Target.host is typed as str), the None check should be removed. As-is, a type checker would flag the unreachable None branch, or a caller could pass None without the type checker warning them.
| infos = socket.getaddrinfo(host, None) | |
| def _is_unsafe_destination(host: str | None) -> bool: |
| A list of resolved IP address strings. Empty when the host cannot be | ||
| parsed or resolved. | ||
| """ | ||
| try: |
There was a problem hiding this comment.
[Bug / Critical] socket.getaddrinfo can also raise OSError (e.g., for system-level network failures, socket.herror, or resource exhaustion). Only catching socket.gaierror means other socket/OS errors will propagate as unhandled exceptions and could crash the scanning engine. Consider catching OSError instead (which is the parent of socket.gaierror and socket.herror), or at minimum also catch OSError alongside socket.gaierror.
| try: | |
| except OSError: |
| assert vulnerabilities == [] | ||
|
|
||
|
|
||
| def test_isUnsafeDestination_coversPrivateAndMetadataRanges() -> None: |
There was a problem hiding this comment.
[Style] Two convention violations here:
-
Testing a private method: This test directly calls
_ip_is_unsafe()which is a private function (prefixed with_). The coding conventions state "Avoid testing or mocking private methods (those starting with an underscore)". Instead, test this behavior through the public API (accept/check), which the other tests in this file already do comprehensively. -
Test naming: This test uses
test_isUnsafeDestination_coversPrivateAndMetadataRangeswith underscores, which doesn't follow thetestAction_conditionCamelCase_expectedResultconvention used by all other tests in this file. It should be renamed to something liketestIpIsUnsafe_whenGivenPrivateAndMetadataAddresses_returnsTrue.
Since the existing tests already cover all the same IPs through the public API, this test can likely be removed entirely.
|
|
||
| Args: | ||
| target: The target to check. | ||
|
|
There was a problem hiding this comment.
[Performance] Beyond the double DNS resolution issue noted on accept(), this guard check is redundant when accept() has already returned True for the same target. The exploit framework guarantees that check() is only called after accept() returns True. If you want to keep this as a safety net, consider documenting the intentional re-resolution (and the TOCTOU implications), or pass a pre-computed flag from accept to check to avoid the duplicate work.
| # Safety net: re-validate destination even though accept() already verified it. | |
| # NOTE: This causes a second DNS resolution and introduces TOCTOU risk — | |
| # the DNS record could change between accept() and check(). In practice the | |
| # framework guarantees check() is only called after accept() returns True, | |
| # so this guard is defensive only. | |
| if _is_unsafe_destination(target.host) is False: | |
| return [] |
Summary
Adds a generic, CWE-918 based detection rule (no CVE) to Agent Asteroid that surfaces Missing SSRF Destination Validation on scan-target URLs.
Agent Asteroid is the downstream scanning-engine consumer that performs the real HTTP fetch of scan targets. When an upstream scan orchestrator (e.g. the reporting engine's
WebScanMutation/Copilotcreate_web_scan_tool/RescanMutation) forwards attacker-controlled target URLs to the engine without validating the destination, the engine is asked to fetch private, loopback, link-local, reserved or cloud-metadata destinations — turning the scanning infrastructure into a blind SSRF relay. This rule implements the defense-in-depth control recommended in the ticket: it detects and reports those unsafe destinations on the fetch side instead of performing the request.Behaviour
accept(target)returnsTrueonly when the target host resolves to a private / loopback / link-local / reserved / multicast / unspecified IP, a known cloud-metadata hostname (metadata.google.internal,169.254.169.254, ...), or the AWS IPv6 IMDS networkfd00:ec2::/16.check(target)reports a single MEDIUMServer-Side Request Forgery (SSRF) Destination Validation Bypassvulnerability and never performs the HTTP fetch (reporting the destination is the detection signal; fetching it would realize the SSRF).socket.getaddrinfo) is used so public-looking domains that resolve to private IPs (DNS rebinding) are caught.Changes
agent/exploits/ssrf_destination_validation.pyregistered viaexploits_registry.tests/exploits/ssrf_destination_validation_test.pycovering private IPv4, loopback, cloud-metadata IP, metadata hostname, DNS-rebinding domain, AWS IPv6 IMDS, public-target non-reporting, unresolvable hostname non-reporting, and a unit test for the unsafe-destination helper covering every payload from the ticket (10.0.0.1,192.168.1.1,127.0.0.1,169.254.169.254,fd00:ec2::254).Verification
ruff format --check .andruff check .pass.mypy agent/exploits/ssrf_destination_validation.pypasses (remaining repo mypy errors are pre-existing missing optional deps).Fixes ticket https://report.ostorlab.co/o/os/remediation/tickets/os-35114