| """Applicability and redacted result normalization for scoped protocol profiles.""" |
| import json |
| import re |
| import tempfile |
| from pathlib import Path |
| from urllib.parse import urljoin, urlsplit |
|
|
| from .module_inputs import ModuleInputs |
| from .pipeline import execute |
| from .web_modules import alert, read_report, same_origin |
| from .protocol_runner import bucket_target |
|
|
| MODULES = {"graphw00f", "corsy", "takeover", "s3-review", "websocket", "vcs-artifacts"} |
| DEADLINES = {"graphw00f": 100, "corsy": 40, "takeover": 35, "s3-review": 30, "websocket": 25, "vcs-artifacts": 35} |
| |
| |
| POLICY_STOPS = {"redirect_not_followed", "scope_blocked", "request_budget", "response_limit"} |
|
|
| LABELS = { |
| "vcs_metadata": ("Exposed VCS metadata candidate", "Medium"), |
| "graphql_present": ("GraphQL metadata endpoint observed", "Informational"), |
| "graphql_introspection": ("GraphQL limited introspection accessible", "Informational"), |
| "graphql_engine": ("GraphQL native engine fingerprint", "Informational"), |
| "cors_observation": ("CORS policy observation", "Informational"), |
| "takeover_candidate": ("Potential dangling service binding", "Medium"), |
| "s3_anonymous_list": ("S3 anonymous ListBucket accepted", "Medium"), |
| "s3_acl_readable": ("S3 bucket ACL anonymously readable", "Informational"), |
| "s3_public_acl": ("S3 public ACL grant advertised", "Medium"), |
| "websocket_present": ("WebSocket opening handshake verified", "Informational"), |
| "websocket_cross_origin": ("Anonymous WebSocket accepts cross-origin handshake", "Informational"), |
| } |
| KINDS = {"vcs-artifacts": {"vcs_metadata"}, "graphw00f": {"graphql_present", "graphql_introspection", "graphql_engine"}, |
| "corsy": {"cors_observation"}, "takeover": {"takeover_candidate"}, |
| "s3-review": {"s3_anonymous_list", "s3_acl_readable", "s3_public_acl"}, |
| "websocket": {"websocket_present", "websocket_cross_origin"}} |
| CORS_CLASSES = {"origin reflected", "post-domain wildcard", "pre-domain wildcard", "null origin allowed", |
| "wildcard value", "third party allowed", "unrecognized underscore", "broken parser", |
| "unescaped regex", "http origin allowed", "invalid value"} |
| REASONS = { |
| "directory_url_required": "Requires a scoped directory URL ending in / without query parameters", |
| "graphql_unconfirmed": "No GraphQL __typename metadata response was confirmed", |
| "s3_root_required": "Requires an explicitly scoped AWS S3 REST bucket-root URL", |
| "websocket_unconfirmed": "No valid RFC 6455 opening handshake; Socket.IO negotiation is outside this profile", |
| "scope_blocked": "A request outside this profile was blocked", |
| "request_budget": "Request budget reached", "deadline": "Profile deadline reached", |
| "request_failed": "A scoped HTTP request failed", "rate_limited": "HTTP 429; profile stopped without retries", |
| "redirect_not_followed": "Redirect not followed; review the exact endpoint URL", |
| "response_limit": "Response sample exceeded 64 KiB", "dns_inconclusive": "DNS evidence was inconclusive", |
| "engine_error": "Native profile error; raw diagnostics redacted", |
| } |
|
|
|
|
| def target_for(module, primary, discovered, values): |
| field = {"graphw00f": "graphql_path", "corsy": "cors_path", "websocket": "websocket_path"}.get(module) |
| explicit = getattr(values, field, "") if field else "" |
| if explicit: |
| return urljoin(urlsplit(primary)._replace(path="/", query="", fragment="").geturl(), explicit) |
| if module in ("graphw00f", "websocket"): |
| pattern = (r"/(?:graphql|gql)(?:/|$)" if module == "graphw00f" |
| else r"/(?:ws|websocket|socket)(?:/|$)") |
| for candidate in [primary] + list(discovered or []): |
| if (isinstance(candidate, str) and same_origin(candidate, primary) |
| and re.search(pattern, urlsplit(candidate).path, re.I)): |
| |
| |
| return urlsplit(candidate)._replace(query="", fragment="").geturl() |
| return None |
| if module == "vcs-artifacts" and (urlsplit(primary).query or urlsplit(primary).path and not urlsplit(primary).path.endswith("/")): |
| return None |
| if module == "s3-review" and bucket_target(primary) is None: |
| return None |
| return primary |
|
|
|
|
| def normalize(module, report, target): |
| findings, seen = [], set() |
| for item in report.get("observations", []): |
| if not isinstance(item, dict) or item.get("kind") not in KINDS[module]: |
| continue |
| kind = item["kind"] |
| name, risk = LABELS[kind] |
| detail = "" |
| if kind == "vcs_metadata": |
| artifact = item.get("artifact") |
| if artifact not in ("git_head", "git_config", "hg_requires", "svn_entries"): |
| continue |
| name += ": " + artifact.replace("_", " ") |
| detail = "Structured " + artifact.replace("_", " ") + " metadata matched, distinct from a random missing-path response. No repository reconstruction, object/history download or credential extraction. File contents are redacted." |
| elif kind == "graphql_engine": |
| engine = item.get("engine") |
| if not isinstance(engine, str) or not re.fullmatch(r"[a-z0-9_-]{1,50}", engine): |
| continue |
| detail = "Native engine candidate: " + engine + ". Signatures can be ambiguous." |
| elif kind.startswith("graphql_"): |
| detail = "Metadata-only observation; introspection is not itself a vulnerability. No schema reconstruction, application data, mutations or subscriptions." |
| elif kind == "cors_observation": |
| category = item.get("category") |
| if category not in CORS_CLASSES: |
| continue |
| detail = "Native Corsy category: " + category + ". " |
| name += ": " + category |
| if category in {"origin reflected", "post-domain wildcard", "pre-domain wildcard", "null origin allowed"} and item.get("credential_candidate") is True: |
| risk = "Medium" |
| detail += "Credential-enabled policy candidate; authenticated response sensitivity and browser exploitability require manual validation." |
| else: |
| detail += "No credential-enabled browser exposure established. Wildcard ACAO with ACAC does not allow credentialed browser reads." |
| elif kind == "takeover_candidate": |
| provider = item.get("provider") |
| evidence = item.get("evidence") |
| if not isinstance(provider, str) or not re.fullmatch(r"[A-Za-z0-9 /()._-]{1,100}", provider) or evidence not in ("cname_and_nxdomain", "cname_and_literal_fingerprint"): |
| continue |
| name += ": " + provider |
| detail = provider + "; evidence: " + evidence.replace("_", " ") + ". Candidate only; ownership verification/claimability is not proven. No service registration or binding attempted." |
| elif kind == "s3_public_acl": |
| permission = item.get("permission") |
| if permission not in ("READ", "WRITE", "READ_ACP", "WRITE_ACP", "FULL_CONTROL"): |
| continue |
| name += ": " + permission |
| risk = "High" if permission in ("WRITE", "WRITE_ACP", "FULL_CONTROL") else "Medium" |
| detail = "AllUsers ACL advertises " + permission + ". Metadata only; no write/delete was attempted and effective policy enforcement is not proven." |
| elif kind.startswith("s3_"): |
| detail = "Anonymous bucket-root metadata check. Public access may be intentional. No object contents, object names or owner identifiers retained." |
| else: |
| detail = "Opening handshake only; no frames, fuzzing or supplied credentials. Cross-origin anonymous acceptance alone does not prove authenticated WebSocket hijacking." |
| key = (kind, detail) |
| if key in seen: |
| continue |
| seen.add(key) |
| findings.append(alert(module, name, target, detail + " Raw responses and native context are redacted.", risk, kind)) |
| return findings[:100] |
|
|
|
|
| async def run(module, url, host, agent, delay, discovered=None, inputs=None): |
| values = inputs if isinstance(inputs, ModuleInputs) else ModuleInputs(**(inputs or {})) |
| target = target_for(module, url, discovered, values) |
| if target is None: |
| reason = ("Requires an explicit same-origin endpoint path or a matching discovered URL" |
| if module in ("graphw00f", "websocket") else REASONS["directory_url_required"] if module == "vcs-artifacts" else REASONS["s3_root_required"]) |
| return {"source": module, "target": url, "status": "skipped", "alerts": [], "reason": reason, "output": reason} |
| with tempfile.TemporaryDirectory(prefix="asfb-" + module + "-") as tmp: |
| spec, output = Path(tmp) / "spec.json", Path(tmp) / "report.json" |
| spec.write_text(json.dumps({"module": module, "target": target, "host": host, "agent": agent, "delay": delay}), encoding="utf-8") |
| rows, status = await execute(["/opt/protocol/bin/python", "/app/app/protocol_runner.py", str(spec), str(output)], DEADLINES[module]) |
| report = read_report(output) |
| if not isinstance(report, dict) or report.get("profile") != module or not isinstance(report.get("observations"), list): |
| return {"source": module, "target": target, "status": status if status != "completed" else "failed", |
| "alerts": [], "reason": "No valid bounded protocol report; native diagnostics redacted", |
| "output": module + ": incomplete; no clean-scan claim"} |
| findings = normalize(module, report, target) |
| early_stop = (status == "completed" and report.get("status") == "partial" |
| and report.get("stop_reason") in POLICY_STOPS) |
| if early_stop: |
| status = "findings" if findings else "completed" |
| if not early_stop and status == "completed" and report.get("status") in ("skipped", "partial", "failed"): |
| status = report["status"] |
| if not early_stop and status == "completed" and report.get("status") != "completed": |
| status = "partial" |
| if status == "completed" and findings: |
| status = "findings" |
| requests = report.get("requests", 0) |
| requests = requests if type(requests) is int and 0 <= requests <= 64 else 0 |
| metadata = {"requests": requests, "observations": len(findings)} |
| for key in ("excluded_probe_types", "provider_candidates"): |
| value = report.get(key) |
| if type(value) is int and 0 <= value <= 100: |
| metadata[key] = value |
| aliases = report.get("dns_aliases") |
| if isinstance(aliases, list): |
| metadata["dns_aliases"] = [name for name in aliases if isinstance(name, str) and re.fullmatch(r"[a-z0-9_.-]{1,253}", name)][:4] |
| reason = REASONS.get(report.get("stop_reason"), "") |
| if not reason: |
| reason = {"graphw00f": "Native query-only fingerprinting and limited introspection; no schema reconstruction", |
| "corsy": "Native Corsy tests; supplied account credentials are not used", |
| "takeover": "Pinned provider data with CNAME corroboration; candidates require manual claimability review", |
| "s3-review": "S3 anonymous ListBucket max-keys=0 and ACL metadata; no object downloads or writes", |
| "websocket": "Two opening handshakes at most; no application frames or authenticated hijack testing", |
| "vcs-artifacts": "Five GETs: missing-path control and four structured VCS metadata checks; no reconstruction or history"}[module] |
| if early_stop: |
| reason = "Early stop: " + reason + "; policy enforced; coverage is incomplete" |
| confirmed_graphql = module == "graphw00f" and any(item["rule_id"] == "graphql_present" for item in findings) |
| return {"source": module, "target": target, "status": status, "alerts": findings, "metadata": metadata, |
| "requests": requests, "deadline_seconds": DEADLINES[module], "reason": reason, |
| **({"early_stop": True, "stop_reason": report["stop_reason"], "coverage_complete": False} |
| if early_stop else {}), |
| "endpoints": [target] if confirmed_graphql else [], |
| "output": f"{module}: {len(findings)} observations; {requests} scoped HTTP requests; status: {status}; native context redacted"} |
|
|
