diff --git a/scripts/v2/workers/diagnosis/observations.py b/scripts/v2/workers/diagnosis/observations.py new file mode 100644 index 00000000..45375136 --- /dev/null +++ b/scripts/v2/workers/diagnosis/observations.py @@ -0,0 +1,555 @@ +"""Bounded host-only Resolver/NFM reads; DNS window matches remain hypotheses.""" +import datetime as dt +import ipaddress +import json +import math +import os +import re +import time + +from botocore.exceptions import BotoCoreError, ClientError + +from . import discovery as d +from .scoped_clients import ScopeError + +# Provider/credential/payload failures are source gaps; anything else is a projection defect. +_SOURCE_ERRORS = (ClientError, BotoCoreError, ScopeError, ValueError) + + +class _NetworkBound(TimeoutError): + """The stage deadline or analysis budget: already disclosed as incomplete, never a source gap.""" + +MAX_GROUPS = 8 +MAX_REGIONS = 8 +MAX_PAGES = 2 +MAX_ROWS = 50 +MAX_MONITORS = 2 +MAX_QUERIES = 6 +MAX_POLLS = 8 # Backoff waits up to ~7.75s per query inside the 35s stage deadline. +# A DNS binding pairs a source and database in one VPC, so an INTER_VPC contributor +# can never be that pair; querying it would only spend the bounded query budget. +CATEGORIES = ("INTRA_AZ", "INTER_AZ") +MAX_ANSWERS = 8 +# Resolver query logs carry answers as [{"Rdata", "Type", "Class"}]. Logs Insights +# flattens JSON arrays into indexed fields (answers.0.Rdata), so the bounded query +# projects those explicitly; one extra index proves the answer list was not truncated. +_ANSWER_FIELDS = tuple(f"answers.{i}.{key}" for i in range(MAX_ANSWERS + 1) + for key in ("Type", "Rdata")) +DNS_FIELDS = ("account_id", "region", "vpc_id", "query_timestamp", "query_name", + "rcode", "answers", "srcaddr", "instance", *_ANSWER_FIELDS) +_VPC_ID = re.compile(r"vpc-[0-9a-f]+") +MAX_OBSERVATION_RECORDS = 200 # DNS + flow evidence emitted by one project() call + + +def dns_query(vpcs): + """Selective bounded query: only validated, associated VPCs and answered NOERROR rows.""" + vpcs = sorted(vpcs) + if not vpcs or any(not isinstance(v, str) or not _VPC_ID.fullmatch(v) for v in vpcs): + raise ValueError("dns_query_vpc_invalid") + listed = ", ".join(f'"{v}"' for v in vpcs) + return (f'filter vpc_id in [{listed}] and rcode = "NOERROR" and ispresent(answers.0.Rdata) ' + "| fields account_id, region, vpc_id, query_timestamp, query_name, rcode, " + "answers, " + ", ".join(_ANSWER_FIELDS) + + f", srcaddr, srcids.instance as instance | limit {MAX_ROWS}") +_sleep = time.sleep + + +def _ip(value): + try: + return str(ipaddress.ip_address(value)) if isinstance(value, str) else None + except ValueError: + return None + + +def _host(value): + if not isinstance(value, str) or len(value) > 254: + return None + value = value.rstrip(".").lower() + return value if re.fullmatch(r"[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?", value) else None + + +def _qualified(bundle): + account = bundle.get("coverage", {}).get("account") + regions = bundle.get("coverage", {}).get("regions", {}).get("selected", []) + resources = {r.get("id"): r for r in bundle.get("resources", [])} + captures = {} + for item in bundle.get("evidence", []): + stamp = d._stamp(item.get("captured_at")) + if (item.get("kind") == "inventory" and item.get("fresh") is True + and item.get("coverage", {}).get("complete") is True + and item.get("account") == account and stamp and d._fresh(stamp, 300)): + for rid in item.get("resource_ids", []): + resource = resources.get(rid) + facts = resource.get("facts", {}) if resource else {} + recorded = item.get("facts", {}) + keys = ("vpc_id", "private_ip_address", "endpoint_address", "endpoint_port") + if (resource and item.get("region") == resource.get("region") + and isinstance(recorded, dict) and all(recorded.get(key) == facts[key] + for key in keys if key in facts)): + captures[rid] = max(stamp, captures.get(rid, stamp)) + return [(r, captures[r["id"]]) for r in bundle.get("resources", []) + if r.get("id") in captures and r.get("account") == account + and r.get("region") in regions] + + +def _flattened_answers(row): + if not any(key in row for key in _ANSWER_FIELDS): + return None + if any(row.get(f"answers.{MAX_ANSWERS}.{key}") is not None for key in ("Type", "Rdata")): + return [] # more answers than projected: refuse rather than bind a partial set + answers = [] + for i in range(MAX_ANSWERS): + kind, rdata = row.get(f"answers.{i}.Type"), row.get(f"answers.{i}.Rdata") + if kind is None and rdata is None: + break + answers.append({"Type": kind, "Rdata": rdata}) + if any(row.get(f"answers.{j}.{key}") is not None for j in range(len(answers) + 1, MAX_ANSWERS) + for key in ("Type", "Rdata")): + return [] # a gap in the indexes cannot be an exact answer list + return answers + + +def _answers(row): + answers = row.get("answers") + if answers is None or answers == "": + answers = _flattened_answers(row) + if isinstance(answers, str): + if len(answers) > 8192: + return None + try: + answers = json.loads(answers) + except (ValueError, TypeError): + return None + if not isinstance(answers, list) or not 1 <= len(answers) <= 16: + return None + addresses, aliases = set(), set() + for answer in answers: + if not isinstance(answer, dict): + return None + # Real Resolver query log shape: {"Rdata": ..., "Type": ..., "Class": "IN"}. + kind, rdata = answer.get("Type"), answer.get("Rdata") + if answer.get("Class", "IN") != "IN": + return None + if kind in {"A", "AAAA"}: + address = _ip(rdata) + if address is None or (":" in address) != (kind == "AAAA"): + return None + addresses.add(address) + elif kind == "CNAME": + host = _host(rdata) + if host is None: + return None + aliases.add(host) + else: + return None + # Multiple aliases cannot establish the order/ownership of a CNAME chain. + return (addresses, aliases) if addresses and len(aliases) <= 1 else None + + +def _dns_binding(bundle, row, qualified, start, end): + account = bundle["coverage"]["account"] + stamp = d._stamp(row.get("query_timestamp")) + region, vpc = row.get("region"), row.get("vpc_id") + answer = _answers(row) + name, source_ip = _host(row.get("query_name")), _ip(row.get("srcaddr")) + if (row.get("account_id") != account or not stamp or not start <= stamp <= end + or row.get("rcode") != "NOERROR" or not answer or not name or not source_ip): + return None + addresses, aliases = answer + local = [(r, capture) for r, capture in qualified if r["region"] == region + and r.get("facts", {}).get("vpc_id") == vpc + and abs((stamp - capture).total_seconds()) <= 300] + instance = row.get("instance") + sources = [r for r, _ in local if r["kind"] in {"ec2", "ecs_task"} + and _ip(r.get("facts", {}).get("private_ip_address")) == source_ip + and ((r["kind"] == "ec2" and instance == r["resource_id"]) + or (r["kind"] == "ecs_task" and not instance))] + if not sources and any(r["kind"] in {"lambda", "lambda_function"} + and _ip(r.get("facts", {}).get("private_ip_address")) == source_ip for r, _ in local): + d._mark(bundle, "gaps", "lambda_eni_identity_unresolved") + targets = [r for r, _ in local if r["kind"] in {"rds", "rds_instance", "rds_proxy", "rds_cluster"} + and _host(r.get("facts", {}).get("endpoint_address")) in {name, *aliases}] + if len(sources) != 1 or len(targets) != 1: + return None + return {"source": sources[0], "target": targets[0], "stamp": stamp, + "name": name, "ips": addresses, "source_ip": source_ip, "instance": instance, + "region": region, "vpc": vpc, "source_id": row.get("query_source")} + + +def _network_matches(binding, flow, account): + if not isinstance(flow, dict): + return False + for side in ("local", "remote"): + if (flow.get(side + "VpcId") != binding["vpc"] + or flow.get(side + "Region") != binding["region"]): + return False + for suffix, service, native in (("InstanceArn", "ec2", "instance/"), + ("VpcArn", "ec2", "vpc/"), + ("SubnetArn", "ec2", "subnet/")): + arn = flow.get(side + suffix) + if arn: + identifier = flow.get(side + suffix.removesuffix("Arn") + "Id") + if not isinstance(identifier, str) or arn != ( + f"arn:aws:{service}:{binding['region']}:{account}:{native}{identifier}"): + return False + if (flow.get("snatIp") and _ip(flow["snatIp"]) != _ip(flow.get("localIp")) + or flow.get("dnatIp") and _ip(flow["dnatIp"]) != _ip(flow.get("remoteIp"))): + return False + if flow.get("query_region") and flow["query_region"] != binding["region"]: + return False + value = flow.get("value") + if not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(value) or value <= 0: + return False + port = binding["target"]["facts"].get("endpoint_port") + if port is not None and (isinstance(flow.get("targetPort"), bool) or flow.get("targetPort") != port): + return False + for source, target in (("local", "remote"), ("remote", "local")): + if (_ip(flow.get(source + "Ip")) == binding["source_ip"] + and _ip(flow.get(target + "Ip")) in binding["ips"]): + instance = flow.get(source + "InstanceId") + if binding["source"]["kind"] == "ec2": + return instance == binding["instance"] + # Node EC2 identity or Pod metadata does not prove an ECS task owner. + return not instance and not flow.get("kubernetesMetadata") + return False + + +def project(bundle, dns_rows, flow_rows, *, start, end, source_id): + """Project source clocks and narrowly qualified DNS/IP candidates, never calls.""" + start, end = d._stamp(start), d._stamp(end) + if not start or not end or not 0 < (end - start).total_seconds() <= 3600: + d._mark(bundle, "gaps", "network_window_invalid") + return bundle + qualified = _qualified(bundle) + if any(isinstance(row, dict) and row.get("kubernetesMetadata") + for row in flow_rows[:MAX_ROWS * MAX_QUERIES]): + d._mark(bundle, "gaps", "pod_identity_unresolved") + unique = {} + for row in dns_rows[:MAX_ROWS * MAX_GROUPS]: + binding = _dns_binding(bundle, row, qualified, start, end) if isinstance(row, dict) else None + if binding is None: + d._mark(bundle, "gaps", "dns_binding_unresolved") + continue + # Repeated resolutions of one (instance, database, query) add no dependency evidence. + key = (binding["source"]["id"], binding["target"]["id"], binding["name"], + frozenset(binding["ips"])) + if key not in unique or binding["stamp"] > unique[key]["stamp"]: + unique[key] = binding + bindings = list(unique.values()) + peers = {} + for binding in bindings: + for address in binding["ips"]: + key = (binding["region"], binding["vpc"], address) + peers.setdefault(key, set()).add(binding["target"]["id"]) + conflict = {key for key, targets in peers.items() if len(targets) != 1} + if conflict: + d._mark(bundle, "gaps", "dns_ip_conflict") + window = {"start": start.isoformat(), "end": end.isoformat()} + emitted, flow_matched = 0, False + for binding in bindings: + if emitted >= MAX_OBSERVATION_RECORDS: + d._mark(bundle, "incomplete", "observation_cap") + break + resources = [binding["source"]["id"], binding["target"]["id"]] + facts = {"query_name": binding["name"], "dns_timestamp": binding["stamp"].isoformat(), + "answer_ips": sorted(binding["ips"]), "ttl_known": False, + "binding": "dns_window_correlation", "direction": "undirected", + "source_id": d._id("source", binding["source_id"] or "resolver_query_logs")} + evidence = d._evidence(bundle, "resolver_query_logs", "observed_dns_resolution", resources, + facts, binding["stamp"], region=binding["region"], fresh=d._fresh(binding["stamp"], 300), + window=window, coverage={"complete": False, "sampled": True}) + d._relationship(bundle, *resources, "observed_dns_resolution", evidence) + emitted += 1 + if any((binding["region"], binding["vpc"], ip) in conflict for ip in binding["ips"]): + continue + for row in flow_rows[:MAX_ROWS * MAX_QUERIES]: + if not _network_matches(binding, row, bundle["coverage"]["account"]): + continue + flow_matched = True + if emitted >= MAX_OBSERVATION_RECORDS: + d._mark(bundle, "incomplete", "observation_cap") + break + if row.get("unit") not in (None, "Bytes"): + d._mark(bundle, "gaps", "network_unit_unsupported") + continue + item = d._evidence(bundle, "network_flow_monitor", "observed_network_flow", resources, + {**facts, "dns_source_id": facts["source_id"], + "source_id": d._id("source", row.get("query_source") or source_id), "metric_name": "DATA_TRANSFERRED", + "value": row["value"], "unit": "Bytes", "port": row.get("targetPort"), + "destination_category": row.get("destinationCategory")}, + end, region=binding["region"], fresh=d._fresh(end, 300), window=window, + coverage={"complete": False, "sampled": True}) + d._relationship(bundle, *resources, "observed_network_flow", item) + emitted += 1 + if bindings and not flow_matched and any(isinstance(row, dict) for row in flow_rows[:MAX_ROWS * MAX_QUERIES]): + # Contributors were returned but none bound to a resolved pair: not "no traffic". + d._mark(bundle, "gaps", "nfm_rows_unmatched") + return bundle + + +class _Reads: + """One source-stage clock and budget checks before clients and each SDK call.""" + def __init__(self, bundle, budget, factory): + self.bundle, self.budget, self.factory = bundle, budget, factory + self.deadline = time.monotonic() + 35 + self.clients = {} + self.bounded = False + + def available(self): + if self.budget.remaining(analysis=True) <= 2 or time.monotonic() >= self.deadline: + d._mark(self.bundle, "incomplete", "network_evidence_deadline") + self.bounded = True + return False + return True + + def call(self, service, region, operation, **parameters): + if not self.available(): + raise _NetworkBound("network_evidence_deadline") + key = service, region + if key not in self.clients: + self.clients[key] = self.factory(service, region, self.budget) + if not self.available(): + raise _NetworkBound("network_evidence_deadline") + client = self.clients[key] + config = getattr(client, "_client_config", None) + if config is not None: + timeout = config.connect_timeout + config.read_timeout + if min(self.budget.remaining(analysis=True), self.deadline - time.monotonic()) <= timeout: + d._mark(self.bundle, "incomplete", "network_evidence_deadline") + self.bounded = True + raise _NetworkBound("network_evidence_deadline") + return getattr(client, operation)(**parameters) + + def pause(self, attempt=0): + if self.available(): + _sleep(max(0, min(0.25 * 2 ** attempt, 1.5, self.budget.remaining(analysis=True) - 2, + self.deadline - time.monotonic()))) + + def stop(self, service, region, operation, **parameters): + # Cancellation still runs after the stage deadline (an abandoned query keeps scanning), + # but only on an existing client and while the job's own budget leaves room. + client = self.clients.get((service, region)) + if client is None or self.budget.remaining(analysis=True) <= 1: + d._mark(self.bundle, "incomplete", "network_query_cleanup_failed") + return + try: + getattr(client, operation)(**parameters) + except Exception: + d._mark(self.bundle, "incomplete", "network_query_cleanup_failed") + + +def _pages(reads, service, region, operation, key, *, lower=False): + token = None + for _ in range(MAX_PAGES): + page_size = 25 if lower else 50 + params = {"maxResults" if lower else "MaxResults": page_size} + if token: + params["nextToken" if lower else "NextToken"] = token + response = reads.call(service, region, operation, **params) + rows = response.get(key, []) + if not isinstance(rows, list): + raise ValueError("invalid_source_rows") + if len(rows) > page_size: + d._mark(reads.bundle, "incomplete", "network_source_row_cap") + yield from rows[:page_size] + token = response.get("nextToken" if lower else "NextToken") + if not token: + return + d._mark(reads.bundle, "incomplete", "network_source_page_cap") + + +def _groups(account, regions): + try: + values = json.loads(os.environ.get("DIAGNOSIS_DNS_LOG_GROUP_ARNS", "[]")) + except ValueError: + return {}, True + if not isinstance(values, list) or len(values) > MAX_GROUPS: + return {}, True + groups = {} + for value in values: + if not isinstance(value, str): + return {}, True + match = re.fullmatch(r"arn:aws:logs:([a-z0-9-]+):(\d{12}):log-group:([A-Za-z0-9_./#-]{1,512})(?::\*)?", value) + if not match or match[2] != account: + return {}, True + if match[1] not in regions: # another run's region; not a malformed allowlist + continue + groups.setdefault(match[1], {})[value.removesuffix(":*")] = match[3] + return groups, False + + +def _dns(reads, region, approved, vpcs, start, end): + rows = [] + # IAM route53resolver:ListResolverQueryLogConfigs(+Associations below): workers.tf + # worker_diagnosis_network Sid DiagnosisObservationDiscovery (runtime_read_condition). + configs = {row.get("Id"): row.get("DestinationArn", "").removesuffix(":*") + for row in _pages(reads, "route53resolver", region, "list_resolver_query_log_configs", "ResolverQueryLogConfigs") + if isinstance(row, dict) and row.get("OwnerId") == reads.bundle["coverage"]["account"] + and row.get("Status") == "CREATED" and isinstance(row.get("DestinationArn"), str) + and row["DestinationArn"].removesuffix(":*") in approved} + selected = {} + for row in _pages(reads, "route53resolver", region, "list_resolver_query_log_config_associations", "ResolverQueryLogConfigAssociations"): + if (isinstance(row, dict) and row.get("Status") == "ACTIVE" and row.get("ResourceId") in vpcs + and row.get("ResolverQueryLogConfigId") in configs): + selected.setdefault(configs[row["ResolverQueryLogConfigId"]], set()).add(row["ResourceId"]) + for arn in sorted(selected)[:MAX_GROUPS]: + query_id, finished = None, False + try: + # IAM logs:StartQuery: workers.tf worker_diagnosis_network Sid DiagnosisApprovedResolverLogRead (allowlisted groups). + response = reads.call("logs", region, "start_query", logGroupName=approved[arn], + startTime=int(start.timestamp()), endTime=int(end.timestamp()), queryString=dns_query(selected[arn]), limit=MAX_ROWS) + query_id = response.get("queryId") + if not isinstance(query_id, str) or not query_id: + raise ValueError("dns_query_id_missing") + for attempt in range(MAX_POLLS): + # IAM logs:GetQueryResults: workers.tf worker_diagnosis_network Sid DiagnosisQueryResultsAndCancellation. + result = reads.call("logs", region, "get_query_results", queryId=query_id) + status = result.get("status") + if status == "Complete": + finished = True + for fields in result.get("results", [])[:MAX_ROWS]: + if isinstance(fields, list): + row = {f["field"]: f.get("value") for f in fields[:64] + if isinstance(f, dict) and f.get("field") in DNS_FIELDS} + if (row.get("region") == region and row.get("vpc_id") in selected[arn] + and row.get("account_id") == reads.bundle["coverage"]["account"]): + rows.append({**row, "query_source": d._id("query", arn, query_id)}) + else: + d._mark(reads.bundle, "gaps", "dns_group_scope_mismatch") + d._mark(reads.bundle, "incomplete", "dns_logs_bounded_window") + break + if status not in {"Scheduled", "Running"}: + finished = True + d._mark(reads.bundle, "gaps", "dns_query_failed") + break + reads.pause(attempt) + if not finished: + d._mark(reads.bundle, "incomplete", "dns_query_poll_cap") + finally: + if query_id and not finished: + # IAM logs:StopQuery: workers.tf worker_diagnosis_network Sid DiagnosisQueryResultsAndCancellation. + reads.stop("logs", region, "stop_query", queryId=query_id) + return rows + + +def _flows(reads, regions, start, end): + rows, monitors, queries = [], 0, 0 + d._mark(reads.bundle, "incomplete", "nfm_top_contributors_partial") + d._mark(reads.bundle, "incomplete", "nfm_categories_omitted") + account = reads.bundle["coverage"]["account"] + for region in regions: + try: + # IAM networkflowmonitor:ListMonitors: workers.tf worker_diagnosis_network Sid DiagnosisObservationDiscovery. + for monitor in _pages(reads, "networkflowmonitor", region, "list_monitors", "monitors", lower=True): + if not isinstance(monitor, dict): + continue + name = monitor.get("monitorName") + if (monitor.get("monitorStatus") != "ACTIVE" or not isinstance(name, str) + or not re.fullmatch(r"[a-zA-Z0-9_.-]{1,255}", name) + or monitor.get("monitorArn") != f"arn:aws:networkflowmonitor:{region}:{account}:monitor/{name}"): + continue + if monitors >= MAX_MONITORS: + d._mark(reads.bundle, "incomplete", "nfm_monitor_cap") + return rows + monitors += 1 + for category in CATEGORIES: + if queries >= MAX_QUERIES: + d._mark(reads.bundle, "incomplete", "nfm_query_cap") + return rows + queries += 1 + query_id, finished = None, False + try: + # IAM networkflowmonitor:{Start,GetQueryStatus,GetQueryResults,Stop}QueryMonitorTopContributors + # (this and the calls below): workers.tf worker_diagnosis_network Sid DiagnosisMonitorObservationQuery. + response = reads.call("networkflowmonitor", region, "start_query_monitor_top_contributors", + monitorName=name, startTime=start, endTime=end, metricName="DATA_TRANSFERRED", + destinationCategory=category, limit=MAX_ROWS) + query_id = response.get("queryId") + if not isinstance(query_id, str) or not query_id: + raise ValueError("nfm_query_id_missing") + for attempt in range(MAX_POLLS): + result = reads.call("networkflowmonitor", region, "get_query_status_monitor_top_contributors", + monitorName=name, queryId=query_id) + if result.get("status") == "SUCCEEDED": + finished = True + result = reads.call("networkflowmonitor", region, "get_query_results_monitor_top_contributors", + monitorName=name, queryId=query_id, maxResults=MAX_ROWS) + contributors = result.get("topContributors", []) + if isinstance(contributors, list): + rows.extend({**row, "unit": result.get("unit"), "query_region": region, + "query_source": d._id("query", monitor["monitorArn"], query_id, category), + "destinationCategory": category} + for row in contributors[:MAX_ROWS] if isinstance(row, dict)) + if result.get("nextToken") or len(contributors) > MAX_ROWS: + d._mark(reads.bundle, "incomplete", "nfm_row_cap") + break + if result.get("status") not in {"QUEUED", "RUNNING"}: + finished = True + d._mark(reads.bundle, "gaps", "nfm_query_failed") + break + reads.pause(attempt) + if not finished: + d._mark(reads.bundle, "incomplete", "nfm_query_poll_cap") + finally: + if query_id and not finished: + reads.stop("networkflowmonitor", region, "stop_query_monitor_top_contributors", + monitorName=name, queryId=query_id) + except _NetworkBound: + return rows + except _SOURCE_ERRORS: + d._mark(reads.bundle, "gaps", "nfm_unavailable") + except Exception: + d._mark(reads.bundle, "degraded", "nfm_projection_failed") + return rows + if not monitors: + d._mark(reads.bundle, "gaps", "nfm_unavailable") + return rows + + +def collect(bundle, budget, *, scope, client_factory): + account = bundle.get("coverage", {}).get("account") + if (os.environ.get("DIAGNOSIS_EVIDENCE_LOOP_ENABLED") != "true" + or os.environ.get("DIAGNOSIS_NETWORK_EVIDENCE_ENABLED") != "true"): + return bundle + if scope != "self" or account != os.environ.get("DIAGNOSIS_HOST_ACCOUNT_ID") or not re.fullmatch(r"\d{12}", account or ""): + d._mark(bundle, "gaps", "network_scope_unsupported") + return bundle + reads = _Reads(bundle, budget, client_factory) + if not reads.available(): + return bundle + from .workload_identities import collect_proxies + collect_proxies(bundle, budget, client_factory) + qualified = _qualified(bundle) + vpcs = {} + for resource, _ in qualified: + vpc = resource.get("facts", {}).get("vpc_id") + if isinstance(vpc, str) and re.fullmatch(r"vpc-[a-f0-9]{8,17}", vpc): + vpcs.setdefault(resource["region"], set()).add(vpc) + regions = sorted(vpcs)[:MAX_REGIONS] + if len(vpcs) > MAX_REGIONS: + d._mark(bundle, "incomplete", "network_region_cap") + if not regions: + d._mark(bundle, "gaps", "network_inventory_unqualified") + return d._finish(bundle) + approved, invalid = _groups(account, regions) + if invalid: + d._mark(bundle, "gaps", "dns_allowlist_invalid") + end = d._now() + start = end - dt.timedelta(minutes=5) + dns_rows = [] + for region in regions: + if region not in approved: + continue + try: + dns_rows.extend(_dns(reads, region, approved[region], vpcs[region], start, end)) + except _NetworkBound: + break + except _SOURCE_ERRORS: + d._mark(bundle, "gaps", "dns_logs_unavailable") + except Exception: + d._mark(bundle, "degraded", "dns_projection_failed") + if not dns_rows and not reads.bounded: + d._mark(bundle, "gaps", "dns_logs_unavailable") + flows = _flows(reads, regions, start, end) + project(bundle, dns_rows, flows, start=start, end=end, source_id="nfm_top_contributors") + return d._finish(bundle) diff --git a/scripts/v2/workers/diagnosis/supporting.py b/scripts/v2/workers/diagnosis/supporting.py new file mode 100644 index 00000000..45a8c119 --- /dev/null +++ b/scripts/v2/workers/diagnosis/supporting.py @@ -0,0 +1,269 @@ +"""Bounded account summaries supporting all pillars without resource cost attribution.""" +from copy import deepcopy +import datetime as dt +import hashlib +import math +import os +import threading + +import boto3 +from botocore.config import Config +from botocore.exceptions import ClientError, ParamValidationError + +_clients = {} +_lock = threading.Lock() +_MAX_PAGES = 3 +_REGION = os.environ.get("AWS_REGION", "ap-northeast-2") + + +def _client(service, region, budget): + remaining = budget.remaining() + if remaining < 10: + raise TimeoutError("supporting_deadline_exhausted") + timeout = min(15, max(1, int(remaining - 5))) + key = (service, region, timeout) + with _lock: + if key not in _clients: + _clients[key] = boto3.client( + service, region_name=region, config=Config( + connect_timeout=3, read_timeout=timeout, + retries={"total_max_attempts": 1}, max_pool_connections=4)) + return _clients[key] + + +def _numeric(value, *, percent=False): + try: + number = float(value) + except (TypeError, ValueError): + return None + if not math.isfinite(number) or (percent and not 0 <= number <= 100): + return None + return number + + +def _mark(bundle, category, value): + if value not in bundle[category]: + bundle[category].append(value) + + +def _now(): + return dt.datetime.now(dt.timezone.utc) + + +def _cost(client, account, budget, bundle, now): + # Cost Explorer's End is exclusive. On the first day of a month the + # month-to-date window is empty, so summarize the previous full month. + month_start = now.date().replace(day=1) + partial = now.date() != month_start + first = month_start if partial else (month_start - dt.timedelta(days=1)).replace(day=1) + start, end = first.isoformat(), (now.date() if partial else month_start).isoformat() + groups, token, seen, complete = [], None, set(), True + for page_number in range(_MAX_PAGES): + if budget.remaining() < 10: + raise TimeoutError("supporting_deadline_exhausted") + params = {"TimePeriod": {"Start": start, "End": end}, "Granularity": "MONTHLY", + "Metrics": ["UnblendedCost"], + "GroupBy": [{"Type": "DIMENSION", "Key": "SERVICE"}], + "Filter": {"Dimensions": {"Key": "LINKED_ACCOUNT", "Values": [account]}}} + if token: + params["NextPageToken"] = token + # IAM ce:GetCostAndUsage (us-east-1): workers.tf worker_diagnosis Sid DiagnosisDataSourcesReadOnly (runtime_read_condition). + result = client.get_cost_and_usage(**params) + for period in result.get("ResultsByTime", []): + for group in period.get("Groups", []): + metric = group.get("Metrics", {}).get("UnblendedCost", {}) + keys = group.get("Keys", []) + amount = _numeric(metric.get("Amount")) + if not keys or amount is None or not isinstance(metric.get("Unit"), str): + _mark(bundle, "incomplete", "cost_invalid_group") + complete = False + continue + groups.append({"service": str(keys[0])[:160], "amount": amount, + "currency": metric["Unit"][:12], + "estimated": period.get("Estimated") is True}) + token = result.get("NextPageToken") + if not token: + break + if token in seen: + _mark(bundle, "incomplete", "cost_repeated_token") + complete = False + break + seen.add(token) + if page_number == _MAX_PAGES - 1: + _mark(bundle, "incomplete", "cost_page_cap") + complete = False + if len(groups) > 40: + _mark(bundle, "incomplete", "cost_group_cap") + complete = False + currencies = {group["currency"] for group in groups} + if len(currencies) > 1: + _mark(bundle, "gaps", "mixed_currency") + # Keep provider service-level amounts. Never infer resource cost or add currencies. + return ({"cost_groups": groups[:40], "resource_cost": False, "partial_month": partial}, + {"start": start, "end": end}, + {"complete": complete, "returned": min(40, len(groups)), + "total": len(groups), "omitted": max(0, len(groups) - 40)}) + + +_SORT_REJECTED_CODES = frozenset(("InvalidInputException", "ValidationException")) + + +def _posture(client, account, bundle): + query = {"Filters": {"AwsAccountId": [{"Value": account, "Comparison": "EQUALS"}], + "RecordState": [{"Value": "ACTIVE", "Comparison": "EQUALS"}], + "WorkflowStatus": [{"Value": "NEW", "Comparison": "EQUALS"}]}, + "MaxResults": 100} + order = "severity_desc" + try: + try: + # Highest severity first, so a capped page is the worst findings, not an arbitrary + # sample (sources.collect_posture parity); fall back unsorted and say so. + # IAM securityhub:GetFindings (both calls): workers.tf worker_diagnosis Sid DiagnosisDataSourcesReadOnly (runtime_read_condition). + response = client.get_findings( + SortCriteria=[{"Field": "SeverityNormalized", "SortOrder": "desc"}], **query) + except (ClientError, ParamValidationError) as error: + if isinstance(error, ClientError) and error.response.get("Error", {}).get("Code") not in _SORT_REJECTED_CODES: + raise + response = client.get_findings(**query) + order = "unsorted" + _mark(bundle, "incomplete", "posture_unsorted_sample") + except ClientError as error: + # Not subscribed is a steady state, not a failure (sources.collect_posture parity). + if error.response.get("Error", {}).get("Code") != "InvalidAccessException": + raise + return {"enabled": False, "posture_counts": {}}, {"filter": "ACTIVE_NEW"}, {"complete": True, "returned": 0} + rows = response.get("Findings", [])[:100] + counts = {} + for row in rows: + if row.get("AwsAccountId") != account: + _mark(bundle, "incomplete", "posture_scope_mismatch") + continue + severity = row.get("Severity", {}).get("Label") + if severity not in {"INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"}: + severity = "UNKNOWN" + counts[severity] = counts.get(severity, 0) + 1 + more = bool(response.get("NextToken")) + if more: + _mark(bundle, "incomplete", "posture_page_cap") + return ({"enabled": True, "posture_counts": counts, "sampled": True, "continuation": more, "order": order}, + {"filter": "ACTIVE_NEW"}, {"complete": not more, "returned": len(rows), "order": order}) + + +def _changes(client, account, bundle, now): + # IAM cloudtrail:LookupEvents: workers.tf worker_diagnosis Sid DiagnosisDataSourcesReadOnly (runtime_read_condition). + response = client.lookup_events( + StartTime=now - dt.timedelta(hours=24), EndTime=now, MaxResults=50, + LookupAttributes=[{"AttributeKey": "ReadOnly", "AttributeValue": "false"}]) + rows = [] + for event in response.get("Events", [])[:50]: + # No CloudTrailEvent JSON, username, raw resource list or message. + event_time = event.get("EventTime") + if not isinstance(event_time, dt.datetime): + _mark(bundle, "incomplete", "change_time_unknown") + continue + rows.append({"event_name": str(event.get("EventName", ""))[:100], + "event_source": str(event.get("EventSource", ""))[:100], + "event_time": event_time.isoformat()}) + more = bool(response.get("NextToken")) + if more: + _mark(bundle, "incomplete", "changes_page_cap") + return ({"change_events": rows, "sampled": True, "continuation": more}, + {"start": (now - dt.timedelta(hours=24)).isoformat(), "end": now.isoformat()}, + {"complete": not more, "returned": len(rows)}) + + +def _commitment(client, account, budget, bundle, now): + params = {"TimePeriod": {"Start": (now.date() - dt.timedelta(days=30)).isoformat(), + "End": now.date().isoformat()}, + "Filter": {"Dimensions": {"Key": "LINKED_ACCOUNT", "Values": [account]}}} + # IAM ce:GetReservationCoverage/GetSavingsPlansCoverage (below; us-east-1): workers.tf worker_diagnosis Sid DiagnosisDataSourcesReadOnly (runtime_read_condition). + ri = client.get_reservation_coverage(**params) + if budget.remaining() < 10: + raise TimeoutError("supporting_deadline_exhausted") + ri_percent = _numeric(ri.get("Total", {}).get("CoverageHours", {}).get( + "CoverageHoursPercentage"), percent=True) + rows = [{"service": "ReservedInstances", "percentage": ri_percent}] + # MONTHLY rows each carry their own TimePeriod; bounded continuation keeps + # completeness honest instead of treating any first-page token as permanent. + more, token, seen = bool(ri.get("NextPageToken")), None, set() + for page_number in range(_MAX_PAGES): + if budget.remaining() < 10: + raise TimeoutError("supporting_deadline_exhausted") + sp = client.get_savings_plans_coverage(**params, Granularity="MONTHLY", MaxResults=20, + **({"NextToken": token} if token else {})) + for row in sp.get("SavingsPlansCoverages", [])[:20]: + period = row.get("TimePeriod") if isinstance(row.get("TimePeriod"), dict) else {} + rows.append({"service": "SavingsPlans", + "start": str(period.get("Start"))[:32] if period.get("Start") else None, + "end": str(period.get("End"))[:32] if period.get("End") else None, + "percentage": _numeric(row.get("Coverage", {}).get("CoveragePercentage"), + percent=True)}) + token = sp.get("NextToken") + if not token: + break + if token in seen or page_number == _MAX_PAGES - 1: + more = True + break + seen.add(token) + if more: + _mark(bundle, "incomplete", "commitment_page_cap") + return ({"commitment_coverage": rows, "sampled": more}, + {"start": params["TimePeriod"]["Start"], "end": params["TimePeriod"]["End"]}, + {"complete": not more, "returned": len(rows)}) + + +def collect(bundle, budget, *, scope): + result = deepcopy(bundle) + account = result.get("coverage", {}).get("account") + # Applied host identity must match; member reads never use this path. + if scope != "self" or result.get("coverage", {}).get("storage_scope") != "self": + return result + if not account or account != os.environ.get("DIAGNOSIS_HOST_ACCOUNT_ID"): + _mark(result, "gaps", "supporting_host_unknown") # host identity absent/mismatched: all skipped + return result + now = _now() + plans = [ + ("cost", "ce", "us-east-1", _cost), ("posture", "securityhub", _REGION, _posture), + ("what_changed", "cloudtrail", _REGION, _changes), ("commitment", "ce", "us-east-1", _commitment), + ] + for key, service, region, fn in plans: + if budget.remaining() < 10: + _mark(result, "degraded", "supporting_deadline_exhausted") + break + try: + client = _client(service, region, budget) + if key == "cost" or key == "commitment": + facts, window, coverage = fn(client, account, budget, result, now) + elif key == "what_changed": + facts, window, coverage = fn(client, account, result, now) + else: + facts, window, coverage = fn(client, account, result) + regions = result.get("coverage", {}).get("regions") + others = sorted(set(regions.get("selected", []) if isinstance(regions, dict) else []) - {region}) + if service != "ce" and others: # single-region API; never claim scope-wide coverage + coverage = {**coverage, "complete": False, "regions_not_queried": others} + _mark(result, "gaps", ("posture" if key == "posture" else "changes") + "_single_region") + captured = now.isoformat() + record_id = "summary-" + hashlib.sha256( + f"{account}:{key}:{captured}".encode()).hexdigest()[:24] + result["evidence"].append({ + "id": record_id, "source": key, "kind": "scope_summary", + "resource_ids": [], "account": account, + "region": "global" if service == "ce" else region, + "facts": facts, "captured_at": captured, "fresh": True, + "window": window, "coverage": coverage}) + result.setdefault("native_sources", {})[key] = { + "status": "ok" if coverage.get("complete") else "partial", + "window": window, "coverage": coverage} + missing = f"{key}_not_collected" + if missing in result["gaps"]: + result["gaps"].remove(missing) + except TimeoutError: # the shared deadline ran out mid-plan: not a provider failure + _mark(result, "degraded", "supporting_deadline_exhausted") + result.setdefault("native_sources", {})[key] = {"status": "not_collected", "reason": "deadline"} + except Exception: # fixed diagnostics; no provider exception data in reports + _mark(result, "degraded", f"{key}_read_failed") + result.setdefault("native_sources", {})[key] = {"status": "failed"} + # Use the same final byte/citation bound as discovery; no second context budget. + from .discovery import _finish + return _finish(result) diff --git a/scripts/v2/workers/diagnosis/test_observations.py b/scripts/v2/workers/diagnosis/test_observations.py new file mode 100644 index 00000000..32a545a8 --- /dev/null +++ b/scripts/v2/workers/diagnosis/test_observations.py @@ -0,0 +1,531 @@ +"""DNS resolution and aggregate traffic retain scope, source clocks and uncertainty.""" +import copy +import datetime as dt +import json + +import pytest + +from diagnosis import discovery, observations +from diagnosis.evidence import EvidenceIndex +from diagnosis.test_evidence import oid +from diagnosis.test_discovery import ACCOUNT, REGION, NOW, I1, Budget + +START = NOW - dt.timedelta(minutes=5) +VPC = "vpc-0123456789abcdef0" +GROUP = f"arn:aws:logs:{REGION}:{ACCOUNT}:log-group:/resolver/test" + + +def bundle(): + resources = [ + {"id": oid("r-instance"), "account": ACCOUNT, "region": REGION, "kind": "ec2", "resource_id": I1, + "facts": {"private_ip_address": "10.0.1.5", "vpc_id": VPC}}, + {"id": oid("r-db"), "account": ACCOUNT, "region": REGION, "kind": "rds", "resource_id": "orders", + "facts": {"endpoint_address": "orders.example.test", "endpoint_port": 5432, "vpc_id": VPC}}] + evidence = [{"id": oid("e-inventory-" + r["id"]), "source": "inventory", "kind": "inventory", + "account": ACCOUNT, "region": REGION, "resource_ids": [r["id"]], "facts": r["facts"], + "captured_at": (NOW - dt.timedelta(minutes=1)).isoformat(), "fresh": True, + "coverage": {"complete": True}} for r in resources] + return {"resources": resources, "evidence": evidence, "relationships": [], + "coverage": {"account": ACCOUNT, "storage_scope": "self", "regions": {"selected": [REGION]}}, + "gaps": [], "degraded": [], "incomplete": []} + + +def dns(**changes): + row = {"account_id": ACCOUNT, "region": REGION, "vpc_id": VPC, + "query_timestamp": (NOW - dt.timedelta(seconds=20)).isoformat(), + "query_name": "orders.example.test.", "rcode": "NOERROR", "srcaddr": "10.0.1.5", + "instance": I1, "answers": [{"Rdata": "10.0.2.7", "Type": "A", "Class": "IN"}], + "srcport": 60000, "raw_sql": "secret"} + return {**row, **changes} + + +def flow(**changes): + return {"localIp": "10.0.1.5", "localInstanceId": I1, "localVpcId": VPC, + "localRegion": REGION, "remoteIp": "10.0.2.7", "remoteVpcId": VPC, + "remoteRegion": REGION, "targetPort": 5432, "value": 321, **changes} + + +@pytest.fixture(autouse=True) +def scope(monkeypatch): + monkeypatch.setenv("DIAGNOSIS_EVIDENCE_LOOP_ENABLED", "true") + monkeypatch.setenv("DIAGNOSIS_NETWORK_EVIDENCE_ENABLED", "true") + monkeypatch.setenv("DIAGNOSIS_HOST_ACCOUNT_ID", ACCOUNT) + monkeypatch.setenv("DIAGNOSIS_DNS_LOG_GROUP_ARNS", json.dumps([GROUP])) + monkeypatch.setattr(discovery, "_now", lambda: NOW) + monkeypatch.setattr(observations, "_sleep", lambda _: None) + + +def test_out_of_region_dns_group_is_skipped_but_malformed_or_foreign_invalidates(monkeypatch): + other = GROUP.replace(REGION, "eu-west-1") + monkeypatch.setenv("DIAGNOSIS_DNS_LOG_GROUP_ARNS", json.dumps([GROUP, other])) + assert observations._groups(ACCOUNT, [REGION]) == ({REGION: {GROUP: "/resolver/test"}}, False) + for bad in (GROUP.replace(ACCOUNT, "999900001111"), "arn:aws:logs:bad"): + monkeypatch.setenv("DIAGNOSIS_DNS_LOG_GROUP_ARNS", json.dumps([GROUP, bad])) + assert observations._groups(ACCOUNT, [REGION]) == ({}, True) + + +def project(data=None, rows=None, flows=None): + data = copy.deepcopy(data if data is not None else bundle()) + observations.project(data, rows if rows is not None else [dns()], + flows if flows is not None else [flow()], start=START, end=NOW, source_id="monitor:test") + return data + + +def test_dns_plus_flow_preserves_original_windows_and_never_becomes_sql_call(): + data = project() + kinds = {e["kind"] for e in data["evidence"]} + assert {"observed_dns_resolution", "observed_network_flow"} <= kinds + assert "observed_call" not in kinds + edge = next(e for e in data["evidence"] if e["kind"] == "observed_network_flow") + assert edge["facts"]["direction"] == "undirected" + assert edge["facts"]["ttl_known"] is False + assert edge["facts"]["binding"] == "dns_window_correlation" + assert edge["window"] == {"start": START.isoformat(), "end": NOW.isoformat()} + assert edge["coverage"]["complete"] is False and edge["coverage"]["sampled"] is True + assert "srcport" not in json.dumps(edge) and "raw_sql" not in json.dumps(edge) + + +@pytest.mark.parametrize("change", [ + {"account_id": "999900001111"}, {"region": "us-east-1"}, {"vpc_id": "vpc-foreign"}, + {"query_timestamp": (START - dt.timedelta(seconds=1)).isoformat()}, + {"query_timestamp": (NOW + dt.timedelta(seconds=1)).isoformat()}, + {"query_timestamp": "malformed"}, {"srcaddr": "10.0.1.6"}, {"instance": "i-foreign"}, + {"rcode": "NXDOMAIN"}, {"answers": []}, {"answers": [{"Rdata": "invalid", "Type": "A", "Class": "IN"}]}, + {"answers": [{"Rdata": "10.0.2.7", "Type": "A", "Class": "CH"}]}, +]) +def test_bad_dns_cannot_establish_resource_dependencies(change): + data = project(rows=[dns(**change)]) + assert not data["relationships"] + assert data["gaps"] + + +@pytest.mark.parametrize("change", [ + {"localVpcId": "vpc-foreign"}, {"remoteVpcId": "vpc-foreign"}, + {"localRegion": "us-east-1"}, {"remoteRegion": "us-east-1"}, + {"localInstanceId": "i-foreign"}, {"snatIp": "10.0.9.8"}, {"dnatIp": "10.0.9.8"}, + {"targetPort": 3306}, {"value": 0}, {"value": float("nan")}, + {"localInstanceArn": "arn:aws:ec2:ap-northeast-2:999900001111:instance/i-foreign"}, +]) +def test_bad_flow_keeps_dns_but_does_not_bind_network_dependency(change): + data = project(flows=[flow(**change)]) + assert any(e["kind"] == "observed_dns_resolution" for e in data["evidence"]) + assert not any(e["kind"] == "observed_network_flow" for e in data["evidence"]) + + +def test_duplicate_endpoint_and_ip_reuse_are_not_guessed(): + data = bundle() + duplicate = copy.deepcopy(data["resources"][1]); duplicate["id"] = oid("r-other"); duplicate["resource_id"] = "other" + data["resources"].append(duplicate) + item = copy.deepcopy(data["evidence"][1]); item["id"] = oid("e-other"); item["resource_ids"] = [oid("r-other")] + data["evidence"].append(item) + assert not project(data)["relationships"] + duplicate["facts"]["endpoint_address"] = "other.example.test" + data["resources"][-1] = duplicate + data["evidence"][-1]["facts"] = copy.deepcopy(duplicate["facts"]) + reused = project(data, rows=[dns(), dns(query_name="other.example.test")]) + assert not any(e["kind"] == "observed_network_flow" for e in reused["evidence"]) + assert "dns_ip_conflict" in reused["gaps"] + + +def test_stale_or_unqualified_inventory_cannot_establish_identity(): + for change in ({"fresh": False}, {"coverage": {"complete": False}}, + {"captured_at": (NOW - dt.timedelta(hours=2)).isoformat()}): + data = bundle(); data["evidence"][0].update(change) + assert not project(data)["relationships"] + + +def test_context_and_admission_keep_weak_binding_as_hypothesis(): + data = project() + row = next(e for e in data["evidence"] if e["kind"] == "observed_network_flow") + index = EvidenceIndex(data, ACCOUNT) + finding = {"id": "finding-peer", "pillar": "reliability", "severity": "medium", + "resource_ids": [oid("r-instance"), oid("r-db")], "evidence_ids": [row["id"]], + "claim": "observation", "explanation": "The scoped peer is a dependency candidate.", + "prerequisites": [], "verification": "graph_refresh"} + assert index.validate_finding(finding)["claim"] == "hypothesis" + admitted = next(e for e in index.model_context()["evidence"] if e["id"] == row["id"]) + assert admitted["facts"]["ttl_known"] is False + assert admitted["facts"]["direction"] == "undirected" + assert admitted["facts"]["query_name"] == "orders.example.test" + + +# The argname must not shadow the autouse `scope` fixture: its loop gate, allowlist +# and frozen clock must be active so only the parametrized refusal stops the calls. +@pytest.mark.parametrize("gate,report_scope,host,gap", [ + ("false", "self", ACCOUNT, None), ("true", ACCOUNT, ACCOUNT, "network_scope_unsupported"), + ("true", "self", "999900001111", "network_scope_unsupported")]) +def test_gate_or_scope_refusal_makes_no_calls(monkeypatch, gate, report_scope, host, gap): + from diagnosis import workload_identities + assert observations.os.environ["DIAGNOSIS_EVIDENCE_LOOP_ENABLED"] == "true" + monkeypatch.setenv("DIAGNOSIS_NETWORK_EVIDENCE_ENABLED", gate) + monkeypatch.setenv("DIAGNOSIS_HOST_ACCOUNT_ID", host) + monkeypatch.setattr(workload_identities, "collect_proxies", lambda *a: pytest.fail("refusal bypass")) + data = observations.collect(bundle(), Budget(), scope=report_scope, + client_factory=lambda *a: pytest.fail("No SDK call permitted")) + assert not data["relationships"] + assert (gap in data["gaps"]) if gap else not data["gaps"] + + +class Provider: + def __init__(self, *, configs=True, pending=False, next_token=False): + self.calls = []; self.configs = configs; self.pending = pending; self.next_token = next_token + + def list_resolver_query_log_configs(self, **params): + self.calls.append(("configs", params)) + return {"ResolverQueryLogConfigs": [{"Id": "rqlc-test", "OwnerId": ACCOUNT, "Status": "CREATED", + "DestinationArn": GROUP}] if self.configs else []} + + def list_resolver_query_log_config_associations(self, **params): + self.calls.append(("associations", params)) + return {"ResolverQueryLogConfigAssociations": [{"ResolverQueryLogConfigId": "rqlc-test", + "ResourceId": VPC, "Status": "ACTIVE"}]} + + def start_query(self, **params): + self.calls.append(("dns_start", params)); return {"queryId": "dns-test"} + + def get_query_results(self, **params): + self.calls.append(("dns_results", params)) + if self.pending: return {"status": "Running"} + # Logs Insights flattens the JSON answers array into indexed fields. + row = dns() + for i, answer in enumerate(row.pop("answers")): + row[f"answers.{i}.Rdata"], row[f"answers.{i}.Type"] = answer["Rdata"], answer["Type"] + return {"status": "Complete", "results": [[{"field": k, "value": v} for k,v in row.items()] + + [{"field": "@ptr", "value": "opaque"}]]} + + def stop_query(self, **params): + self.calls.append(("dns_stop", params)); return {} + + def list_monitors(self, **params): + self.calls.append(("monitors", params)) + return {"monitors": [{"monitorName": "test", "monitorStatus": "ACTIVE", + "monitorArn": f"arn:aws:networkflowmonitor:{REGION}:{ACCOUNT}:monitor/test"}]} + + def start_query_monitor_top_contributors(self, **params): + self.calls.append(("nfm_start", params)); return {"queryId": "nfm-test"} + + def get_query_status_monitor_top_contributors(self, **params): + self.calls.append(("nfm_status", params)); return {"status": "RUNNING" if self.pending else "SUCCEEDED"} + + def get_query_results_monitor_top_contributors(self, **params): + self.calls.append(("nfm_results", params)) + return {"topContributors": [flow()], "unit": "Bytes", "nextToken": "more" if self.next_token else None} + + def stop_query_monitor_top_contributors(self, **params): + self.calls.append(("nfm_stop", params)); return {} + + +def test_collection_uses_scoped_allowlisted_sources_and_preserves_partial_coverage(monkeypatch): + monkeypatch.setattr(observations, "_sleep", lambda _: None) + fake = Provider(next_token=True) + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert any(e["kind"] == "observed_network_flow" for e in data["evidence"]) + assert len([c for c,_ in fake.calls if c == "nfm_start"]) <= 6 + query = next(p for c,p in fake.calls if c == "dns_start") + assert query["logGroupName"] == "/resolver/test" and "@message" not in query["queryString"] + assert "srcport" not in query["queryString"] + assert "nfm_top_contributors_partial" in data["incomplete"] + + +def test_unapproved_or_unassociated_logs_are_never_queried(monkeypatch): + monkeypatch.setenv("DIAGNOSIS_DNS_LOG_GROUP_ARNS", "[]") + fake = Provider() + observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert not any(c == "dns_start" for c,_ in fake.calls) + + +def test_absent_instrumentation_is_an_explicit_gap(monkeypatch): + fake = Provider(configs=False) + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert "dns_logs_unavailable" in data["gaps"] + assert not data["relationships"] + + +def test_deadline_starts_no_new_query(): + data = observations.collect(bundle(), Budget(1), scope="self", + client_factory=lambda *a: pytest.fail("No SDK call after deadline")) + assert "network_evidence_deadline" in data["incomplete"] + + +def test_pending_queries_are_stopped_at_poll_cap(monkeypatch): + sleeps = [] + monkeypatch.setattr(observations, "_sleep", sleeps.append) + fake = Provider(pending=True) + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + # Logs Insights / NFM need seconds: one query waits ~7.75s, never one long sleep. + assert sum(sleeps[:observations.MAX_POLLS - 1]) >= 5 and max(sleeps) <= 1.5 + assert any(c == "dns_stop" for c,_ in fake.calls) + assert any(c == "nfm_stop" for c,_ in fake.calls) + assert not data["relationships"] and data["incomplete"] + + +def test_emitted_requests_validate_against_offline_botocore_schemas(monkeypatch): + from botocore import xform_name + from botocore.session import Session + from botocore.validate import validate_parameters + session, fake = Session(), Provider() + errors = [] + monkeypatch.setattr(observations, "_sleep", lambda _: None) + + class Client: + def __init__(self, service): + self.model = session.get_service_model(service) + + def __getattr__(self, name): + implementation = getattr(fake, name) + operation = next(op for op in self.model.operation_names + if xform_name(op) == name) + + def call(**params): + try: + shape = self.model.operation_model(operation).input_shape + validate_parameters(params, shape) + # botocore's validator does not enforce every modeled maximum. + for key, value in params.items(): + maximum = shape.members[key].metadata.get("max") + if maximum is not None and isinstance(value, (int, float)): + assert value <= maximum, (operation, key, value, maximum) + except Exception as exc: + errors.append(str(exc)) + raise + return implementation(**params) + return call + + data = observations.collect(bundle(), Budget(), scope="self", + client_factory=lambda service, *a: Client(service)) + assert not errors + assert any(e["kind"] == "observed_network_flow" for e in data["evidence"]) + + +@pytest.mark.parametrize("change", [ + {"localInstanceArn": f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-other"}, + {"localVpcArn": f"arn:aws:ec2:{REGION}:{ACCOUNT}:vpc/vpc-other"}, + {"remoteVpcArn": f"arn:aws:ec2:{REGION}:{ACCOUNT}:vpc/vpc-other"}, +]) +def test_same_account_arn_identifier_conflicts_refuse_binding(change): + data = project(flows=[flow(**change)]) + assert not any(e["kind"] == "observed_network_flow" for e in data["evidence"]) + + +def test_dns_rows_are_checked_against_their_own_group_association(monkeypatch): + data = bundle() + for r in data["resources"]: + r["facts"]["vpc_id"] = "vpc-fedcba9876543210" + subnet = {"id": oid("r-subnet"), "account": ACCOUNT, "region": REGION, + "kind": "subnet", "resource_id": "subnet-test", "facts": {"vpc_id": VPC}} + data["resources"].append(subnet) + data["evidence"].append({**data["evidence"][0], "id": oid("e-subnet"), "resource_ids": [subnet["id"]], + "facts": subnet["facts"]}) + + class WrongVpc(Provider): + def get_query_results(self, **params): + result = super().get_query_results(**params) + for field in result["results"][0]: + if field["field"] == "vpc_id": field["value"] = "vpc-fedcba9876543210" + return result + + fake = WrongVpc() + observed = observations.collect(data, Budget(), scope="self", client_factory=lambda *a: fake) + assert not any(e["kind"] == "observed_dns_resolution" for e in observed["evidence"]) + + +def test_inventory_binding_requires_recorded_address_not_only_a_fresh_flag(): + data = bundle() + data["evidence"][0]["facts"] = {"private_ip_address": "10.0.1.6", "vpc_id": VPC} + assert not project(data)["relationships"] + + +def test_cached_sdk_timeout_cannot_exceed_remaining_query_budget(monkeypatch): + from botocore.config import Config + fake = Provider() + fake._client_config = Config(connect_timeout=1, read_timeout=10) + monkeypatch.setattr(observations, "_sleep", lambda _: None) + observations.collect(bundle(), Budget(5), scope="self", client_factory=lambda *a: fake) + assert not any(c in {"dns_start", "nfm_start"} for c,_ in fake.calls) + + +def test_evidence_loop_gate_off_invokes_neither_sdk_nor_proxy_helper(monkeypatch): + from diagnosis import workload_identities + monkeypatch.setenv("DIAGNOSIS_EVIDENCE_LOOP_ENABLED", "false") + monkeypatch.setattr(workload_identities, "collect_proxies", lambda *a: pytest.fail("gate bypass")) + observations.collect(bundle(), Budget(), scope="self", + client_factory=lambda *a: pytest.fail("gate bypass")) + + +def test_sources_share_exact_window_derived_from_one_end_clock(monkeypatch): + from diagnosis import workload_identities + calls = [] + + def advancing(): + calls.append(True) + return NOW + dt.timedelta(seconds=len(calls)) + + monkeypatch.setattr(discovery, "_now", advancing) + monkeypatch.setattr(workload_identities, "collect_proxies", lambda *a: None) + fake = Provider() + observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + dns_query = next(p for c,p in fake.calls if c == "dns_start") + assert dns_query["endTime"] - dns_query["startTime"] == 300 + for kind, params in fake.calls: + if kind == "nfm_start": + assert (params["endTime"] - params["startTime"]).total_seconds() == 300 + assert int(params["startTime"].timestamp()) == dns_query["startTime"] + + +def test_nfm_global_monitor_and_query_caps_are_visible(monkeypatch): + class ManyMonitors(Provider): + def list_monitors(self, **params): + self.calls.append(("monitors", params)) + return {"monitors": [{"monitorName": f"m{i}", "monitorStatus": "ACTIVE", + "monitorArn": f"arn:aws:networkflowmonitor:{REGION}:{ACCOUNT}:monitor/m{i}"} + for i in range(3)]} + + fake = ManyMonitors() + monkeypatch.setattr(observations, "_sleep", lambda _: None) + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert len([c for c,_ in fake.calls if c == "nfm_start"]) == 2 * len(observations.CATEGORIES) + assert "nfm_monitor_cap" in data["incomplete"] + + +def test_reversed_peer_flow_is_undirected_and_cname_is_not_call_evidence(): + data = project(rows=[dns(query_name="alias.example.test", answers=[ + {"Rdata": "orders.example.test", "Type": "CNAME", "Class": "IN"}, + {"Rdata": "10.0.2.7", "Type": "A", "Class": "IN"}])], flows=[flow(localIp="10.0.2.7", + remoteIp="10.0.1.5", localInstanceId=None, remoteInstanceId=I1)]) + assert any(e["kind"] == "observed_network_flow" for e in data["evidence"]) + assert not any(e["kind"] == "observed_call" for e in data["evidence"]) + + +def test_malformed_admitted_fact_shape_cannot_raise_or_confirm_a_call(): + data = project() + row = next(e for e in data["evidence"] if e["kind"] == "observed_network_flow") + row["facts"] = ["untrusted"] + index = EvidenceIndex(data, ACCOUNT) + finding = index.validate_finding({"id": "finding-shape", "pillar": "reliability", + "severity": "medium", "resource_ids": [oid("r-instance"), oid("r-db")], "evidence_ids": [row["id"]], + "claim": "observation", "explanation": "The peer needs verification.", + "prerequisites": [], "verification": "graph_refresh"}) + assert finding["claim"] == "hypothesis" + + +def test_inventory_scope_and_source_provenance_survive_without_raw_metadata(): + data = bundle() + data["evidence"][0]["region"] = "us-east-1" + assert not project(data)["relationships"] + fake = Provider() + collected = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + admitted = EvidenceIndex(collected, ACCOUNT).model_context() + dns_row = next(e for e in admitted["evidence"] if e["kind"] == "observed_dns_resolution") + network_row = next(e for e in admitted["evidence"] if e["kind"] == "observed_network_flow") + assert dns_row["facts"]["source_id"].startswith("source:") + assert network_row["facts"]["dns_source_id"] == dns_row["facts"]["source_id"] + assert network_row["facts"]["source_id"] != dns_row["facts"]["source_id"] + assert GROUP not in json.dumps(admitted) and "raw_sql" not in json.dumps(admitted) + + +def test_node_metadata_and_shared_lambda_addresses_keep_explicit_identity_gaps(): + data = bundle() + data["resources"][0]["kind"] = "lambda_function" + result = project(data, rows=[dns(instance=None)], flows=[flow(kubernetesMetadata={"localPodName": "p"})]) + assert not result["relationships"] + assert "lambda_eni_identity_unresolved" in result["gaps"] + assert "pod_identity_unresolved" in result["gaps"] + + +def test_resolver_answers_parse_real_shape_from_list_json_and_flattened_fields(): + real = [{"Rdata": "10.0.2.7", "Type": "A", "Class": "IN"}] + flattened = dns(); flattened.pop("answers") + flattened.update({"answers.0.Rdata": "10.0.2.7", "answers.0.Type": "A"}) + for row in (dns(answers=real), dns(answers=json.dumps(real)), flattened): + assert observations._answers(row) == ({"10.0.2.7"}, set()) + # The legacy guessed keys are not the Resolver log shape. + assert observations._answers(dns(answers=[{"answer_type": "A", "rdata": "10.0.2.7"}])) is None + + +def test_flattened_answers_beyond_projection_or_with_gaps_are_refused(): + overflow = dns(); overflow.pop("answers") + for i in range(observations.MAX_ANSWERS + 1): + overflow[f"answers.{i}.Rdata"], overflow[f"answers.{i}.Type"] = "10.0.2.7", "A" + gap = dns(); gap.pop("answers") + gap.update({"answers.0.Rdata": "10.0.2.7", "answers.0.Type": "A", + "answers.2.Rdata": "10.0.2.8", "answers.2.Type": "A"}) + assert observations._answers(overflow) is None + assert observations._answers(gap) is None + query = observations.dns_query([VPC]) + assert "answers.0.Rdata" in query and f"answers.{observations.MAX_ANSWERS}.Type" in query + + +def test_dns_query_is_selective_and_validates_vpc_ids(): + query = observations.dns_query({VPC}) + assert f'vpc_id in ["{VPC}"]' in query and 'rcode = "NOERROR"' in query + assert "ispresent(answers.0.Rdata)" in query and query.index("filter") < query.index("limit") + for bad in ([], ['vpc-0"] or rcode = "x'], ["vpc-XYZ"]): + with pytest.raises(ValueError): + observations.dns_query(bad) + fake = Provider() + observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + sent = next(p for c, p in fake.calls if c == "dns_start")["queryString"] + assert sent == observations.dns_query({VPC}) + + +def test_inter_vpc_is_not_queried_and_unmatched_rows_are_disclosed(): + assert "INTER_VPC" not in observations.CATEGORIES + data = project(flows=[flow(remoteVpcId="vpc-0fedcba9876543210")]) + assert not any(e["kind"] == "observed_network_flow" for e in data["evidence"]) + assert "nfm_rows_unmatched" in data["gaps"] + assert "nfm_rows_unmatched" not in project()["gaps"] + + +def test_duplicate_dns_rows_dedupe_and_correlation_output_is_capped(monkeypatch): + repeated = [dns(query_timestamp=(NOW - dt.timedelta(seconds=s)).isoformat()) for s in range(10, 60)] + data = project(rows=repeated, flows=[flow(value=v) for v in range(1, 40)]) + dns_rows = [e for e in data["evidence"] if e["kind"] == "observed_dns_resolution"] + assert len(dns_rows) == 1 and dns_rows[0]["facts"]["dns_timestamp"] == (NOW - dt.timedelta(seconds=10)).isoformat() + monkeypatch.setattr(observations, "MAX_OBSERVATION_RECORDS", 5) + capped = project(flows=[flow(value=v) for v in range(1, 40)]) + assert len([e for e in capped["evidence"] if e["kind"].startswith("observed_")]) == 5 + assert "observation_cap" in capped["incomplete"] + + +def test_facts_less_inventory_rows_do_not_crash_dns_binding(): + data = bundle() + bare = {"id": oid("r-bare"), "account": ACCOUNT, "region": REGION, "kind": "ec2", "resource_id": "i-bare"} + data["resources"].append(bare) + data["evidence"].append({**data["evidence"][0], "id": oid("e-bare"), "resource_ids": [bare["id"]], "facts": {}}) + assert not project(data, rows=[dns(vpc_id=None)])["relationships"] + + +def test_stage_deadline_is_incomplete_only_and_still_stops_started_queries(monkeypatch): + monkeypatch.setattr(observations, "_sleep", lambda _: None) + clock = [0.0] + monkeypatch.setattr(observations.time, "monotonic", lambda: clock[0]) + fake = Provider(pending=True) + original = fake.get_query_results + + def slow(**params): + clock[0] += 100 # the 35 s stage deadline passes while the DNS query is polled + return original(**params) + fake.get_query_results = slow + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert "network_evidence_deadline" in data["incomplete"] + assert "dns_logs_unavailable" not in data["gaps"] and "nfm_unavailable" not in data["gaps"] + assert any(c == "dns_stop" for c, _ in fake.calls) + + +def test_provider_error_is_a_gap_and_projection_defect_is_degraded(monkeypatch): + from botocore.exceptions import ClientError + monkeypatch.setattr(observations, "_sleep", lambda _: None) + fake = Provider() + + def denied(**params): + raise ClientError({"Error": {"Code": "AccessDeniedException", "Message": "no"}}, "ListMonitors") + fake.list_monitors = denied + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert "nfm_unavailable" in data["gaps"] and "nfm_projection_failed" not in data["degraded"] + + fake = Provider() + fake.list_monitors = lambda **params: {"monitors": [{"monitorName": "test", "monitorStatus": "ACTIVE", + "monitorArn": f"arn:aws:networkflowmonitor:{REGION}:{ACCOUNT}:monitor/test"}]} + fake.start_query_monitor_top_contributors = lambda **params: (_ for _ in ()).throw(KeyError("defect")) + data = observations.collect(bundle(), Budget(), scope="self", client_factory=lambda *a: fake) + assert "nfm_projection_failed" in data["degraded"] diff --git a/scripts/v2/workers/diagnosis/test_supporting.py b/scripts/v2/workers/diagnosis/test_supporting.py new file mode 100644 index 00000000..ebb77fc0 --- /dev/null +++ b/scripts/v2/workers/diagnosis/test_supporting.py @@ -0,0 +1,242 @@ +"""Native account summaries retain scope, currency and bounded coverage.""" +import datetime as dt +import pytest + +ACCOUNT = "111122223333" + + +class Budget: + def remaining(self, analysis=True): + return 1000 + + +def base(): + return {"resources": [], "evidence": [], "relationships": [], + "coverage": {"account": ACCOUNT, "storage_scope": "self"}, + "gaps": [], "degraded": [], "incomplete": [], + "native_sources": {k: {"status": "not_collected"} for k in + ("cost", "posture", "what_changed", "commitment", "idle", "datasources_obs")}} + + +class Client: + def get_cost_and_usage(self, **kwargs): + assert kwargs["Filter"]["Dimensions"]["Values"] == [ACCOUNT] + return {"ResultsByTime": [{"Estimated": True, "Groups": [ + {"Keys": ["Amazon EC2"], "Metrics": {"UnblendedCost": {"Amount": "14.5", "Unit": "USD"}}}, + {"Keys": ["Amazon RDS"], "Metrics": {"UnblendedCost": {"Amount": "10", "Unit": "EUR"}}}, + ]}]} + + def get_findings(self, **kwargs): + assert kwargs["Filters"]["AwsAccountId"][0]["Value"] == ACCOUNT + return {"Findings": [{"AwsAccountId": ACCOUNT, "Severity": {"Label": "HIGH"}}], + "NextToken": "more"} + + def lookup_events(self, **kwargs): + return {"Events": [{"EventName": "ModifyDBInstance", "EventSource": "rds.amazonaws.com", + "EventTime": dt.datetime.now(dt.timezone.utc), + "Username": "private-user", "CloudTrailEvent": "secret-raw-log"}]} + + def get_reservation_coverage(self, **kwargs): + return {"Total": {"CoverageHours": {"CoverageHoursPercentage": "75.0"}}} + + def get_savings_plans_coverage(self, **kwargs): + return {"SavingsPlansCoverages": [{"Coverage": {"CoveragePercentage": "60.0"}}]} + + +@pytest.fixture +def providers(monkeypatch): + from diagnosis import supporting + monkeypatch.setenv("DIAGNOSIS_HOST_ACCOUNT_ID", ACCOUNT) + monkeypatch.setattr(supporting, "_client", lambda *a: Client()) + # Mid-month clock: the month-to-date window is non-empty on every real run date. + monkeypatch.setattr(supporting, "_now", lambda: dt.datetime(2026, 11, 6, 9, tzinfo=dt.timezone.utc)) + return supporting + + +def test_supporting_sources_are_host_scoped_and_never_leak_raw_events(providers): + result = providers.collect(base(), Budget(), scope="self") + assert {e["source"] for e in result["evidence"]} == {"cost", "posture", "what_changed", "commitment"} + assert all(e["account"] == ACCOUNT and e["resource_ids"] == [] for e in result["evidence"]) + import json + text = json.dumps(result) + assert "private-user" not in text and "secret-raw-log" not in text + assert "ModifyDBInstance" in text + assert "posture_page_cap" in result["incomplete"] + assert result["native_sources"]["cost"]["status"] == "ok" + assert not {"posture_single_region", "changes_single_region"} & set(result["gaps"]) + data = base() + data["coverage"]["regions"] = {"selected": [providers._REGION, "us-west-2"]} + result = providers.collect(data, Budget(), scope="self") + assert {"posture_single_region", "changes_single_region"} <= set(result["gaps"]) + changes = result["native_sources"]["what_changed"] + assert changes["status"] == "partial" and changes["coverage"]["regions_not_queried"] == ["us-west-2"] + assert result["native_sources"]["cost"]["status"] == "ok" + + +def test_mixed_currency_costs_stay_per_service_and_partial_month(providers): + result = providers.collect(base(), Budget(), scope="self") + cost = next(e for e in result["evidence"] if e["source"] == "cost")["facts"] + assert {g["currency"] for g in cost["cost_groups"]} == {"USD", "EUR"} + assert cost["resource_cost"] is False and cost["partial_month"] is True + assert "total_savings" not in cost and "mixed_currency" in result["gaps"] + + +def test_member_scope_and_wrong_host_account_never_use_native_clients(providers, monkeypatch): + monkeypatch.setattr(providers, "_client", lambda *a: pytest.fail("host read occurred")) + member = base() + member["coverage"].update(account="999900001111", storage_scope="999900001111") + assert not providers.collect(member, Budget(), scope="999900001111")["evidence"] + assert not providers.collect(member, Budget(), scope="self")["evidence"] + + +def test_unsubscribed_security_hub_is_steady_state_not_degraded(providers, monkeypatch): + from botocore.exceptions import ClientError + + class Unsubscribed(Client): + def get_findings(self, **kwargs): + raise ClientError({"Error": {"Code": "InvalidAccessException"}}, "GetFindings") + monkeypatch.setattr(providers, "_client", lambda *a: Unsubscribed()) + result = providers.collect(base(), Budget(), scope="self") + posture = next(e for e in result["evidence"] if e["source"] == "posture") + assert posture["facts"]["enabled"] is False and posture["fresh"] is True + assert "posture_read_failed" not in result["degraded"] + assert result["native_sources"]["posture"]["status"] == "ok" + + +def test_posture_sample_is_highest_severity_first_with_unsorted_fallback_disclosed(providers, monkeypatch): + from botocore.exceptions import ClientError + calls = [] + + class Sorted(Client): + def get_findings(self, **kwargs): + calls.append(kwargs) + return super().get_findings(**kwargs) + monkeypatch.setattr(providers, "_client", lambda *a: Sorted()) + result = providers.collect(base(), Budget(), scope="self") + assert calls[0]["SortCriteria"] == [{"Field": "SeverityNormalized", "SortOrder": "desc"}] + posture = next(e for e in result["evidence"] if e["source"] == "posture") + assert posture["facts"]["order"] == "severity_desc" + + class Rejects(Client): + def get_findings(self, **kwargs): + if "SortCriteria" in kwargs: + raise ClientError({"Error": {"Code": "InvalidInputException"}}, "GetFindings") + return super().get_findings(**kwargs) + monkeypatch.setattr(providers, "_client", lambda *a: Rejects()) + result = providers.collect(base(), Budget(), scope="self") + posture = next(e for e in result["evidence"] if e["source"] == "posture") + assert posture["facts"]["order"] == "unsorted" and "posture_unsorted_sample" in result["incomplete"] + + +def test_deadline_mid_plan_is_not_reported_as_a_provider_failure(providers, monkeypatch): + class Slow(Client): + def get_findings(self, **kwargs): + raise TimeoutError("supporting_deadline_exhausted") + monkeypatch.setattr(providers, "_client", lambda *a: Slow()) + result = providers.collect(base(), Budget(), scope="self") + assert "posture_read_failed" not in result["degraded"] + assert result["native_sources"]["posture"] == {"status": "not_collected", "reason": "deadline"} + + +def test_repeated_cost_page_tokens_stop_at_bound(providers, monkeypatch): + class Pages(Client): + def get_cost_and_usage(self, **kwargs): + return {"ResultsByTime": [], "NextPageToken": "repeat"} + monkeypatch.setattr(providers, "_client", lambda *a: Pages()) + result = providers.collect(base(), Budget(), scope="self") + assert "cost_repeated_token" in result["incomplete"] + + +def test_deadline_exhaustion_skips_all_provider_calls(providers, monkeypatch): + class Expired: + def remaining(self, analysis=True): + return 0 + monkeypatch.setattr(providers, "_client", lambda *a: pytest.fail("expired call")) + result = providers.collect(base(), Expired(), scope="self") + assert not result["evidence"] and "supporting_deadline_exhausted" in result["degraded"] + + +def test_all_native_requests_match_locked_sdk_shapes(providers, monkeypatch): + import boto3 + from botocore.validate import validate_parameters + + session = boto3.Session(aws_access_key_id="offline", aws_secret_access_key="offline") + + class ShapeChecked(Client): + def __init__(self, service): + self.model = session.client(service, region_name="ap-northeast-2").meta.service_model + + def __getattribute__(self, name): + value = super().__getattribute__(name) + if name.startswith(("get_", "lookup_")) and callable(value): + def checked(**kwargs): + operation = "".join(part.title() for part in name.split("_")) + validate_parameters(kwargs, self.model.operation_model(operation).input_shape) + return value(**kwargs) + return checked + return value + + monkeypatch.setattr(providers, "_client", lambda service, *a: ShapeChecked(service)) + result = providers.collect(base(), Budget(), scope="self") + assert all(result["native_sources"][key]["status"] in {"ok", "partial"} + for key in ("cost", "posture", "what_changed", "commitment")) + assert not result["degraded"] + + +def test_first_day_of_month_summarizes_previous_full_month(providers, monkeypatch): + windows = [] + + class Recording(Client): + def get_cost_and_usage(self, **kwargs): + windows.append(kwargs["TimePeriod"]) + return super().get_cost_and_usage(**kwargs) + + monkeypatch.setattr(providers, "_client", lambda *a: Recording()) + monkeypatch.setattr(providers, "_now", lambda: dt.datetime(2026, 12, 1, 0, 5, tzinfo=dt.timezone.utc)) + result = providers.collect(base(), Budget(), scope="self") + assert windows == [{"Start": "2026-11-01", "End": "2026-12-01"}] + assert "cost_read_failed" not in result["degraded"] + cost = next(e for e in result["evidence"] if e["source"] == "cost") + assert cost["facts"]["partial_month"] is False + assert cost["window"] == {"start": "2026-11-01", "end": "2026-12-01"} + assert result["native_sources"]["cost"]["status"] == "ok" + # January 1 rolls back across the year boundary. + monkeypatch.setattr(providers, "_now", lambda: dt.datetime(2027, 1, 1, tzinfo=dt.timezone.utc)) + providers.collect(base(), Budget(), scope="self") + assert windows[-1] == {"Start": "2026-12-01", "End": "2027-01-01"} + + +def test_savings_plans_coverage_is_monthly_period_labelled_and_paginated(providers, monkeypatch): + calls = [] + + class Paged(Client): + def get_savings_plans_coverage(self, **kwargs): + calls.append(kwargs) + assert kwargs["Granularity"] == "MONTHLY" + month = "10" if "NextToken" not in kwargs else "11" + page = {"SavingsPlansCoverages": [{ + "TimePeriod": {"Start": f"2026-{month}-01", "End": f"2026-{month}-28"}, + "Coverage": {"CoveragePercentage": "60.0"}}]} + return {**page, "NextToken": "p2"} if "NextToken" not in kwargs else page + + monkeypatch.setattr(providers, "_client", lambda *a: Paged()) + result = providers.collect(base(), Budget(), scope="self") + commitment = next(e for e in result["evidence"] if e["source"] == "commitment") + plans = [r for r in commitment["facts"]["commitment_coverage"] if r["service"] == "SavingsPlans"] + assert [(r["start"], r["end"]) for r in plans] == [("2026-10-01", "2026-10-28"), ("2026-11-01", "2026-11-28")] + assert [c.get("NextToken") for c in calls] == [None, "p2"] + assert commitment["coverage"]["complete"] is True and "commitment_page_cap" not in result["incomplete"] + + +def test_absent_or_mismatched_host_identity_marks_the_skip(providers, monkeypatch): + monkeypatch.setattr(providers, "_client", lambda *a: pytest.fail("host read occurred")) + for value in (None, "999900001111"): + if value is None: + monkeypatch.delenv("DIAGNOSIS_HOST_ACCOUNT_ID") + else: + monkeypatch.setenv("DIAGNOSIS_HOST_ACCOUNT_ID", value) + result = providers.collect(base(), Budget(), scope="self") + assert "supporting_host_unknown" in result["gaps"] and not result["evidence"] + member = base() + member["coverage"].update(account="999900001111", storage_scope="999900001111") + assert "supporting_host_unknown" not in providers.collect(member, Budget(), scope="999900001111")["gaps"]