diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 7aa0b838..1820bec8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -298,6 +298,27 @@ days-old graph cannot look confident. `risk:plan` states `plan-time` coverage with its own caveat: it reads the plan, not the graph. Failures degrade the statement, never the prediction. +## Event-triggered recompute + +Escalations used to appear only on the policy evaluator's schedule, so a new +admin grant or trust could sit unevaluated for hours. The incremental +collector now signals when a batch could have moved one: any IAM node +(`IamRole`, `IamUser`, `IamGroup`, `IamPolicy*`, `AccessKey`) or IAM edge +(`TRUSTS`, `HAS_IAM_ROLE`, `ATTACHED_TO`, `MEMBER_OF`, `HAS_ACCESS_KEY`) makes +the batch recompute-worthy, and the collector sends one message to the +`recompute` FIFO queue (`RECOMPUTE_QUEUE_URL`; best-effort — a missing queue or +send failure never fails collection). Two bounds keep the heaviest job in the +system from running constantly: the producer's FIFO deduplication id collapses +a five-minute burst to one message, and the recompute worker — a Deployment +that long-polls the queue — refuses to evaluate more often than +`RECOMPUTE_MIN_INTERVAL_SECONDS` (default 900), deleting messages that arrive +inside the interval as covered by the run that just happened. It runs the +**policy evaluator** only, so escalation paths (and the likelihood on any path +through them) refresh within one incremental cycle; the hourly rule run is what +raises the corresponding issues. A failed evaluation leaves its messages for +SQS redelivery and the queue's DLQ rather than deleting them. The CDK path is +not wired (the scheduled evaluator still covers it). + ## Compliance reconciliation and the trust center The compliance job evaluates the CIS AWS Foundations v3.0.0 and Khalifa diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d490f20..2a0616bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ All notable changes to Khalifa. +## [Unreleased] - Event-triggered recompute + +### Added + +- **Event-triggered policy evaluation**: the incremental collector now signals when a batch could have moved an escalation — any IAM node (`IamRole`, `IamUser`, `IamGroup`, `IamPolicy*`, `AccessKey`) or IAM edge (`TRUSTS`, `HAS_IAM_ROLE`, `ATTACHED_TO`, `MEMBER_OF`, `HAS_ACCESS_KEY`) marks the batch recompute-worthy — sending one best-effort message to a new `recompute` FIFO queue (`RECOMPUTE_QUEUE_URL`); a missing queue or a send failure never fails collection. The `recompute-worker` runtime command runs as a Deployment (rendered only when the queue is wired; policy-evaluator IRSA role plus receive/delete on the queue) that long-polls it and re-runs **the policy evaluator**, so escalation paths refresh within one incremental cycle instead of at the evaluator's scheduled hour; the hourly rule run still raises the corresponding issues. Two bounds keep the heavy job in check: the producer's FIFO deduplication id collapses a five-minute burst to one message, and the worker refuses to evaluate more often than `RECOMPUTE_MIN_INTERVAL_SECONDS` (default 900, chart `recomputeWorker.minIntervalSeconds`), deleting covered messages. A failed evaluation leaves messages for SQS redelivery and the queue's DLQ; the scheduled evaluator remains the backstop. Terraform creates the queue + DLQ, grants the incremental collector `SendMessage` and the policy-evaluator role receive/delete, and flows `recomputeQueueUrl` into the chart. The CDK path is not wired (documented). + +### Notes + +- Still open from the re-review (`~/Documents/Testing/plans/2026-10-27-predictive-completion.md`): freshness tempering inside scores (G9), observed-reachability v2 (remote endpoints, per-hop citation), shift-left v2 (IAM diff/Helm, CI wiring), and the long tail (UEBA breadth, EPSS/KEV, trust-center projections, per-issue risk window/SLO). + ## [Unreleased] - Outcome labels and coverage statements ### Added diff --git a/OPERATIONAL.md b/OPERATIONAL.md index 15ae73ed..b88fa6e3 100644 --- a/OPERATIONAL.md +++ b/OPERATIONAL.md @@ -14,9 +14,10 @@ 8. **cloudtrail-analyzer** - CloudTrail usage CronJob (Athena → AccessAnalyzerCache; optional anomaly nodes via graph-writer) 9. **flow-log-analyzer** - VPC flow-log CronJob (Athena → TrafficObservations; **suspended by default**, reuses the cloudtrail-analyzer IRSA role) 10. **threat-intel-import** - CronJob: imports customer threat-intel lists from S3 into ThreatIndicators (**suspended by default**, runs as the rule-runner service account) -11. **dspm-scanner** - CronJob: secret scanning of S3 objects and Lambda environment variables (Macie primary, regex fallback; findings written via graph-writer) -12. **policy-evaluator / least-privilege-sweep** - Lambdas: effective permissions, escalation paths, rightsizing -13. **IRSA roles** - IAM roles for pod authentication to Neptune/DynamoDB/SQS +11. **recompute-worker** - Deployment (rendered only when the recompute queue is wired): drains IAM-change signals and re-runs the policy evaluator, at most once per `RECOMPUTE_MIN_INTERVAL_SECONDS`; uses the policy-evaluator IRSA role +12. **dspm-scanner** - CronJob: secret scanning of S3 objects and Lambda environment variables (Macie primary, regex fallback; findings written via graph-writer) +13. **policy-evaluator / least-privilege-sweep** - Lambdas: effective permissions, escalation paths, rightsizing +14. **IRSA roles** - IAM roles for pod authentication to Neptune/DynamoDB/SQS ### Network Flow @@ -225,6 +226,21 @@ kubectl create job --from=cronjob/khalifa-rule-runner rule-runner-manual -n secu already consume the findings). Auto-ticketing stays off for matched issues until `/rules/quality` shows measured precision. +### Runbook: Event-triggered recompute + +The recompute worker is rendered only when `collection.recomputeQueueUrl` is set +(the complete example/Terraform wires it). It is always-on by design: +- Producer: the incremental collector sends one message per five-minute window + when a batch touched IAM. Watch `Queued an event-triggered recompute` in the + incremental collector logs. +- Worker: watch `Event-triggered policy evaluation complete` and + `Recompute skipped inside the minimum interval` in the recompute-worker pod. + Tune `recomputeWorker.minIntervalSeconds` (default 900) if the evaluator runs + too often for the graph's size. +- Failures: an evaluation error leaves messages for SQS redelivery; three + receives land them in the recompute DLQ. The scheduled evaluator CronJob + remains the backstop if the worker is down. + ## I. Graph Query Profiling Neptune serves id lookups from the index; a property filter on a whole label diff --git a/README.md b/README.md index fa581340..1fe1890e 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ One member role. Evaluation stays in-account. | **CIEM** | IAM decomposition, effective permissions, escalation paths (assume chains and `iam:PassRole` to compute, RULE-034), CloudTrail-grounded rightsizing. | | **Least privilege** | Sweep of roles and users. Deployable policy recommendations. | | **Secrets** | Secrets Manager and SSM parameter inventory with who-can-read analysis from identity and resource policies, rotation and access age, sharing detection (RULE-010, RULE-032, RULE-033), and admin users holding active access keys (RULE-031). | -| **Risk** | Scheduled rules and attack-path traversals on the live graph (34 rules), with an exploitability factor that raises scores for paths carrying Amazon Inspector findings, an explainable likelihood on every path, a per-fix simulation that counts what each edge removal would clear, matching against customer-supplied threat-intel lists (additive score bonus, TTL-fresh), curated MITRE ATT&CK techniques on every finding, graph-only adversary scenarios that re-run technique chains over the live graph, and a per-rule quality report (suppression vs resolution) that calibrates which rules to trust. | +| **Risk** | Scheduled rules and attack-path traversals on the live graph (34 rules), with an exploitability factor that raises scores for paths carrying Amazon Inspector findings, an explainable likelihood on every path, escalation paths recomputed within minutes of an IAM change (event-triggered and rate-bounded), a per-fix simulation that counts what each edge removal would clear, matching against customer-supplied threat-intel lists (additive score bonus, TTL-fresh), curated MITRE ATT&CK techniques on every finding, graph-only adversary scenarios that re-run technique chains over the live graph, and a per-rule quality report (suppression vs resolution) that calibrates which rules to trust. | | **Detection** | GuardDuty detector coverage per region (including `MISSING`), Security Hub findings joined to internet-exposed resources (RULE-017, RULE-018), CloudTrail anomaly signals: root activity, secret reads from IPs outside each principal's own 30-day baseline, AccessDenied deviations from each principal's own daily baseline (RULE-020 to RULE-022), canary credential tripwires (RULE-023), secret scanning of Lambda environment variables plus IMDSv1-with-role detection (RULE-024), and Amazon Inspector CVEs on internet-exposed workloads (RULE-003). | | **Exposure** | Public EKS control planes (RULE-019) plus public S3/RDS and world-open security groups, derived from collected endpoint and network config, with optional observed traffic from VPC flow logs confirming where traffic actually goes. | | **Compliance** | CIS AWS Foundations v3.0 and the Khalifa baseline, per-control evidence, mapped SOC 2 / ISO 27001 / PCI DSS / NIST 800-53 requirements, and CIS reconciliation against Security Hub before anything is publishable. | @@ -163,6 +163,8 @@ EKS uses [`deploy/helm/khalifa/values.yaml`](deploy/helm/khalifa/values.yaml). T | `THREAT_INTEL_IP_LIST_S3_URI` | s3://bucket/key of the newline-separated IP indicator list (empty disables) | — | | `THREAT_INTEL_ACCOUNT_LIST_S3_URI`| s3://bucket/key of the newline-separated account-id list (empty disables) | — | | `THREAT_INTEL_TTL_DAYS` | Days an imported indicator stays valid (DynamoDB TTL) | `90` | +| `RECOMPUTE_QUEUE_URL` | FIFO queue the incremental collector signals and the recompute worker drains | — | +| `RECOMPUTE_MIN_INTERVAL_SECONDS` | Minimum seconds between event-triggered policy evaluations (recompute worker) | `900` | | `ATHENA_DATABASE` | Glue database for CloudTrail | `khalifa_cloudtrail_db` | | `ATHENA_WORKGROUP` | Athena workgroup | `khalifa-cloudtrail-analysis` | | `CLOUDTRAIL_S3_LOCATION` | CloudTrail S3 prefix | `s3://cloudtrail-logs/AWSLogs/` | @@ -277,7 +279,7 @@ reported as certified. | Path | Contents | | ---------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- | -| [`lambdas/`](lambdas) | Collector, incremental, graph-writer, policy-evaluator, CloudTrail, flow-log, threat-intel import, least-privilege, DSPM, runtime | +| [`lambdas/`](lambdas) | Collector, incremental, graph-writer, policy-evaluator, CloudTrail, flow-log, threat-intel import, recompute, least-privilege, DSPM, runtime | | [`packages/risk-engine`](packages/risk-engine) | Rules, scoring, compliance | | [`api-service/`](api-service) · [`ui/`](ui) | REST API and Next.js UI | | [`deploy/`](deploy) | Helm chart, Terraform modules, fixtures | diff --git a/deploy/helm/khalifa/templates/configmap.yaml b/deploy/helm/khalifa/templates/configmap.yaml index b51a8355..2aa20b02 100644 --- a/deploy/helm/khalifa/templates/configmap.yaml +++ b/deploy/helm/khalifa/templates/configmap.yaml @@ -75,5 +75,6 @@ data: GRAPH_WRITER_TIMEOUT_MS: {{ .Values.collection.graphWriterTimeoutMs | quote }} GRAPH_WRITER_URL: {{ printf "http://%s-graph-writer:%v" (include "khalifa.fullname" .) .Values.graphWriter.service.port | quote }} SQS_QUEUE_URL: {{ .Values.collection.incrementalQueueUrl | quote }} + RECOMPUTE_QUEUE_URL: {{ .Values.collection.recomputeQueueUrl | quote }} GUARDDUTY_ADMINISTRATOR_ACCOUNT_ID: {{ .Values.collection.guardDutyAdministratorAccountId | default "" | quote }} GUARDDUTY_FINDING_REGIONS: {{ join "," (.Values.collection.guardDutyFindingRegions | default list) | quote }} diff --git a/deploy/helm/khalifa/templates/networkpolicies.yaml b/deploy/helm/khalifa/templates/networkpolicies.yaml index 8efcc44d..fb239ffe 100644 --- a/deploy/helm/khalifa/templates/networkpolicies.yaml +++ b/deploy/helm/khalifa/templates/networkpolicies.yaml @@ -1,5 +1,5 @@ {{- if .Values.networkPolicies.enabled }} -{{- range $component := list "api" "ui" "rule-runner" "org-collector" "incremental" "graph-writer" "compliance" "cloudtrail-analyzer" "flow-log-analyzer" "threat-intel-import" "dspm-scanner" "policy-evaluator" "least-privilege-sweep" }} +{{- range $component := list "api" "ui" "rule-runner" "org-collector" "incremental" "graph-writer" "compliance" "cloudtrail-analyzer" "flow-log-analyzer" "threat-intel-import" "recompute-worker" "dspm-scanner" "policy-evaluator" "least-privilege-sweep" }} apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: diff --git a/deploy/helm/khalifa/templates/recompute-worker.yaml b/deploy/helm/khalifa/templates/recompute-worker.yaml new file mode 100644 index 00000000..61aace32 --- /dev/null +++ b/deploy/helm/khalifa/templates/recompute-worker.yaml @@ -0,0 +1,63 @@ +{{- if .Values.collection.recomputeQueueUrl }} +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "khalifa.fullname" . }}-recompute-worker + labels: + {{- include "khalifa.labels" . | nindent 4 }} + app.kubernetes.io/component: recompute-worker +spec: + replicas: {{ .Values.recomputeWorker.replicas }} + selector: + matchLabels: + {{- include "khalifa.selectorLabels" (dict "root" . "component" "recompute-worker") | nindent 6 }} + template: + metadata: + labels: + {{- include "khalifa.selectorLabels" (dict "root" . "component" "recompute-worker") | nindent 8 }} + {{- with .Values.recomputeWorker.podLabels }} + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.recomputeWorker.podAnnotations }} + annotations: + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + # Reuses the policy-evaluator IRSA role: the worker only runs that + # evaluation, and the role gains receive/delete rights on the queue. + serviceAccountName: {{ include "khalifa.serviceAccountName" (dict "root" . "key" "policyEvaluator" "suffix" "policy-evaluator") }} + automountServiceAccountToken: {{ .Values.serviceAccounts.policyEvaluator.automountServiceAccountToken }} + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + securityContext: + {{- include "khalifa.podSecurityContext" . | nindent 8 }} + {{- if .Values.topologySpread.enabled }} + topologySpreadConstraints: + {{- include "khalifa.topologySpreadConstraints" (dict "root" . "component" "recompute-worker") | nindent 8 }} + {{- end }} + containers: + - name: recompute-worker + image: {{ required "recomputeWorker.image must be an immutable repository@sha256 digest" .Values.recomputeWorker.image | quote }} + imagePullPolicy: IfNotPresent + args: ["recompute-worker"] + securityContext: + {{- include "khalifa.containerSecurityContext" . | nindent 12 }} + env: + - name: RECOMPUTE_MIN_INTERVAL_SECONDS + value: {{ .Values.recomputeWorker.minIntervalSeconds | quote }} + envFrom: + - configMapRef: + name: {{ include "khalifa.fullname" . }}-config + - configMapRef: + name: {{ include "khalifa.fullname" . }}-collection + resources: + {{- toYaml .Values.recomputeWorker.resources | nindent 12 }} + volumeMounts: + - name: tmp + mountPath: /tmp + volumes: + - name: tmp + emptyDir: {} +{{- end }} diff --git a/deploy/helm/khalifa/values.schema.json b/deploy/helm/khalifa/values.schema.json index c45f426b..5f99199b 100644 --- a/deploy/helm/khalifa/values.schema.json +++ b/deploy/helm/khalifa/values.schema.json @@ -16,6 +16,7 @@ "graphWriter", "compliance", "reconciliation", + "recomputeWorker", "threatIntelImport", "flowLogAnalyzer", "cloudtrailAnalyzer", @@ -198,6 +199,7 @@ "trustCenterSyncClientId": { "type": "string", "pattern": "^(|[0-9a-f]{32}\\.access)$" } } }, + "recomputeWorker": { "$ref": "#/definitions/recomputeWorker" }, "threatIntelImport": { "$ref": "#/definitions/cronJob" }, "flowLogAnalyzer": { "$ref": "#/definitions/cronJob" }, "cloudtrailAnalyzer": { "$ref": "#/definitions/cronJob" }, @@ -519,6 +521,26 @@ "days": { "type": "integer", "minimum": 1, "maximum": 31 } } }, + "recomputeWorker": { + "type": "object", + "additionalProperties": false, + "required": [ + "image", + "minIntervalSeconds", + "replicas", + "podAnnotations", + "podLabels", + "resources" + ], + "properties": { + "image": { "$ref": "#/definitions/image" }, + "minIntervalSeconds": { "type": "integer", "minimum": 0, "maximum": 86400 }, + "replicas": { "type": "integer", "minimum": 1 }, + "podAnnotations": { "$ref": "#/definitions/stringMap" }, + "podLabels": { "$ref": "#/definitions/stringMap" }, + "resources": { "$ref": "#/definitions/resources" } + } + }, "collection": { "type": "object", "additionalProperties": false, @@ -531,7 +553,8 @@ "maxAttempts", "retryDelayMs", "graphWriterTimeoutMs", - "incrementalQueueUrl" + "incrementalQueueUrl", + "recomputeQueueUrl" ], "properties": { "homeAccountId": { "type": "string", "pattern": "^$|^[0-9]{12}$" }, @@ -566,6 +589,10 @@ "type": "string", "pattern": "^$|^https://sqs\\.[a-z0-9-]+\\.amazonaws\\.com/[0-9]{12}/[A-Za-z0-9_.-]+$" }, + "recomputeQueueUrl": { + "type": "string", + "pattern": "^$|^https://sqs\\.[a-z0-9-]+\\.amazonaws\\.com/[0-9]{12}/[A-Za-z0-9_.-]+$" + }, "guardDutyAdministratorAccountId": { "type": "string", "pattern": "^$|^[0-9]{12}$" }, "guardDutyFindingRegions": { "type": "array", @@ -807,6 +834,7 @@ "cloudtrail-analyzer", "flow-log-analyzer", "threat-intel-import", + "recompute-worker", "dspm-scanner", "policy-evaluator", "least-privilege-sweep" diff --git a/deploy/helm/khalifa/values.yaml b/deploy/helm/khalifa/values.yaml index 4782c965..ce2b644c 100644 --- a/deploy/helm/khalifa/values.yaml +++ b/deploy/helm/khalifa/values.yaml @@ -120,6 +120,7 @@ collection: retryDelayMs: 1000 graphWriterTimeoutMs: 30000 incrementalQueueUrl: '' + recomputeQueueUrl: '' # GuardDuty findings are read from the delegated administrator's detectors # after each organization sweep. Empty leaves it off; its member role needs # guardduty:ListFindings and GetFindings first. @@ -310,6 +311,22 @@ compliance: cpu: 1000m memory: 1Gi +recomputeWorker: + image: '' + # Minimum seconds between event-triggered policy evaluations; messages that + # arrive sooner are deleted as covered by the run that just happened. + minIntervalSeconds: 900 + replicas: 1 + podAnnotations: {} + podLabels: {} + resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: 1000m + memory: 1Gi + threatIntelImport: image: '' schedule: '0 5 * * *' diff --git a/deploy/terraform/khalifa-data/iam.tf b/deploy/terraform/khalifa-data/iam.tf index ff016104..ae999791 100644 --- a/deploy/terraform/khalifa-data/iam.tf +++ b/deploy/terraform/khalifa-data/iam.tf @@ -687,6 +687,14 @@ resource "aws_iam_role_policy" "incremental_collector" { ] Resource = [aws_sqs_queue.incremental.arn] }, + { + # Event-triggered recompute: an IAM-sensitive batch signals the + # worker that the policy evaluation is stale. + Sid = "RecomputeQueueSend" + Effect = "Allow" + Action = ["sqs:GetQueueUrl", "sqs:SendMessage"] + Resource = [aws_sqs_queue.recompute.arn] + }, ], local.home_collection_statements ) @@ -716,6 +724,19 @@ resource "aws_iam_role_policy" "policy_evaluator" { ] Resource = [local.neptune_resource_arn] }, + { + # The recompute worker runs as this role and drains the queue. + Sid = "RecomputeQueue" + Effect = "Allow" + Action = [ + "sqs:ChangeMessageVisibility", + "sqs:DeleteMessage", + "sqs:GetQueueAttributes", + "sqs:GetQueueUrl", + "sqs:ReceiveMessage", + ] + Resource = [aws_sqs_queue.recompute.arn] + }, ] }) } diff --git a/deploy/terraform/khalifa-data/messaging.tf b/deploy/terraform/khalifa-data/messaging.tf index 6bfffe26..8fd7bf1c 100644 --- a/deploy/terraform/khalifa-data/messaging.tf +++ b/deploy/terraform/khalifa-data/messaging.tf @@ -52,6 +52,36 @@ resource "aws_sqs_queue_policy" "incremental" { }) } +# Event-triggered recompute. No queue policy: only this account's workload +# roles send (incremental collector) and receive (recompute worker). +resource "aws_sqs_queue" "recompute_dlq" { + name = "${local.name}-recompute-dlq.fifo" + fifo_queue = true + content_based_deduplication = true + kms_master_key_id = aws_kms_key.this.arn + + tags = merge(var.tags, { Name = "${local.name}-recompute-dlq" }) +} + +resource "aws_sqs_queue" "recompute" { + name = "${local.name}-recompute.fifo" + fifo_queue = true + # The producer sets an explicit deduplication id (one per five-minute + # window), so bursts collapse to a single message. + content_based_deduplication = false + kms_master_key_id = aws_kms_key.this.arn + # A policy evaluation can run for minutes; keep messages invisible for it. + visibility_timeout_seconds = 900 + message_retention_seconds = 14400 + + redrive_policy = jsonencode({ + deadLetterTargetArn = aws_sqs_queue.recompute_dlq.arn + maxReceiveCount = 3 + }) + + tags = merge(var.tags, { Name = "${local.name}-recompute" }) +} + resource "aws_sns_topic" "issues" { name = "${local.name}-issues" kms_master_key_id = aws_kms_key.this.arn diff --git a/deploy/terraform/khalifa-data/outputs.tf b/deploy/terraform/khalifa-data/outputs.tf index 33f65566..aad94946 100644 --- a/deploy/terraform/khalifa-data/outputs.tf +++ b/deploy/terraform/khalifa-data/outputs.tf @@ -120,6 +120,7 @@ output "helm_values" { parentIds = var.collection_parent_ids regions = var.collection_regions incrementalQueueUrl = aws_sqs_queue.incremental.url + recomputeQueueUrl = aws_sqs_queue.recompute.url } networkPolicies = { enabled = true diff --git a/deploy/terraform/khalifa-data/tests/data.tftest.hcl b/deploy/terraform/khalifa-data/tests/data.tftest.hcl index 0eca4ba5..d090ae77 100644 --- a/deploy/terraform/khalifa-data/tests/data.tftest.hcl +++ b/deploy/terraform/khalifa-data/tests/data.tftest.hcl @@ -141,8 +141,14 @@ run "data_plane" { } assert { - condition = aws_sqs_queue.incremental.fifo_queue && aws_sqs_queue.incremental_dlq.fifo_queue - error_message = "Incremental collection must use a FIFO queue and DLQ." + condition = ( + aws_sqs_queue.incremental.fifo_queue && + aws_sqs_queue.incremental_dlq.fifo_queue && + aws_sqs_queue.recompute.fifo_queue && + aws_sqs_queue.recompute_dlq.fifo_queue && + !aws_sqs_queue.recompute.content_based_deduplication + ) + error_message = "Incremental collection and event-triggered recompute must use FIFO queues with DLQs; recompute dedups on the producer's five-minute window id." } assert { @@ -232,6 +238,8 @@ run "data_plane" { } assert { + # The recompute worker reuses this role, so the pin covers the queue it + # drains as well as the Neptune access the evaluation needs. condition = toset(flatten([ for statement in jsondecode(aws_iam_role_policy.policy_evaluator.policy).Statement : flatten([statement.Action]) @@ -240,8 +248,13 @@ run "data_plane" { "neptune-db:WriteDataViaQuery", "neptune-db:DeleteDataViaQuery", "neptune-db:GetEngineStatus", + "sqs:ChangeMessageVisibility", + "sqs:DeleteMessage", + "sqs:GetQueueAttributes", + "sqs:GetQueueUrl", + "sqs:ReceiveMessage", ]) - error_message = "The policy evaluator policy must contain exactly the Neptune read/write/delete actions and nothing else." + error_message = "The policy evaluator policy must contain exactly the Neptune read/write/delete actions, the recompute queue drain rights, and nothing else." } assert { @@ -342,6 +355,8 @@ run "data_plane" { condition = alltrue([ aws_sqs_queue.incremental.kms_master_key_id == aws_kms_key.this.arn, aws_sqs_queue.incremental_dlq.kms_master_key_id == aws_kms_key.this.arn, + aws_sqs_queue.recompute.kms_master_key_id == aws_kms_key.this.arn, + aws_sqs_queue.recompute_dlq.kms_master_key_id == aws_kms_key.this.arn, aws_sns_topic.issues.kms_master_key_id == aws_kms_key.this.arn, ]) error_message = "SQS queues and the issues SNS topic must use the Khalifa CMK." diff --git a/deploy/testdata/eks-deployment-values.yaml b/deploy/testdata/eks-deployment-values.yaml index f588d57c..2f20e578 100644 --- a/deploy/testdata/eks-deployment-values.yaml +++ b/deploy/testdata/eks-deployment-values.yaml @@ -65,6 +65,7 @@ collection: regions: - us-west-2 incrementalQueueUrl: https://sqs.us-west-2.amazonaws.com/111122223333/validation-khalifa-incremental.fifo + recomputeQueueUrl: https://sqs.us-west-2.amazonaws.com/111122223333/validation-khalifa-recompute.fifo api: image: 111122223333.dkr.ecr.us-west-2.amazonaws.com/khalifa-api@sha256:1111111111111111111111111111111111111111111111111111111111111111 @@ -133,6 +134,9 @@ threatIntelImport: schedule: '0 5 * * *' timeZone: America/Los_Angeles +recomputeWorker: + image: 111122223333.dkr.ecr.us-west-2.amazonaws.com/khalifa-runtime@sha256:4444444444444444444444444444444444444444444444444444444444444444 + dspmScanner: image: 111122223333.dkr.ecr.us-west-2.amazonaws.com/khalifa-runtime@sha256:4444444444444444444444444444444444444444444444444444444444444444 schedule: '0 4 * * *' diff --git a/jest.config.js b/jest.config.js index 4ba1fc17..62047357 100644 --- a/jest.config.js +++ b/jest.config.js @@ -96,6 +96,14 @@ module.exports = { testPathIgnorePatterns: ['\\.graph\\.test\\.ts$'], modulePathIgnorePatterns: ['/dist/', '/node_modules/'], }, + { + displayName: 'recompute-worker', + rootDir: '/lambdas/recompute-worker', + preset: 'ts-jest', + testEnvironment: 'node', + testMatch: ['/**/*.test.ts'], + modulePathIgnorePatterns: ['/dist/', '/node_modules/'], + }, { displayName: 'runtime', rootDir: '/lambdas/runtime', diff --git a/lambdas/incremental-collector/index.ts b/lambdas/incremental-collector/index.ts index 7779b3f9..c1fda7a7 100644 --- a/lambdas/incremental-collector/index.ts +++ b/lambdas/incremental-collector/index.ts @@ -1,4 +1,9 @@ -import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs'; +import { + SQSClient, + ReceiveMessageCommand, + DeleteMessageCommand, + SendMessageCommand, +} from '@aws-sdk/client-sqs'; import { EC2Client, DescribeSecurityGroupsCommand } from '@aws-sdk/client-ec2'; import { IAMClient, GetUserCommand } from '@aws-sdk/client-iam'; import { OrganizationsClient } from '@aws-sdk/client-organizations'; @@ -12,6 +17,7 @@ import type { GraphNode, GraphEdge } from '../shared/types'; import { Logger } from '../shared/types'; import { invokeGraphWriter } from '../shared/graph-writer-client'; import { getTargetCredentials } from '../shared/aws-credentials'; +import { recomputeDeduplicationId, recomputeReasons } from '../shared/recompute'; const logger = new Logger('incremental-processor'); interface CollectionScope { @@ -116,6 +122,7 @@ export async function processBatch( edges, observedAt: event.detail?.eventTime || event.time || new Date().toISOString(), }); + await signalRecompute(sqsClient, environment, nodes, edges); } await sqsClient.send( @@ -137,6 +144,40 @@ export async function processBatch( export const handler = () => processBatch(); +// Event-triggered recompute: an IAM grant, role, key or trust written by this +// batch can create an escalation the policy evaluator would otherwise only see +// on its schedule, so enqueue a bounded recompute signal. FIFO dedup (one id +// per five-minute window) collapses bursts; the worker's own minimum interval +// bounds the runs. Best-effort by design: a missing queue or a send failure +// never fails collection. +export async function signalRecompute( + sqsClient: SQSClient, + environment: NodeJS.ProcessEnv, + nodes: GraphNode[], + edges: GraphEdge[], + now: Date = new Date() +): Promise { + const queueUrl = environment.RECOMPUTE_QUEUE_URL?.trim(); + if (!queueUrl) return false; + const reasons = recomputeReasons(nodes, edges); + if (reasons.length === 0) return false; + try { + await sqsClient.send( + new SendMessageCommand({ + QueueUrl: queueUrl, + MessageBody: JSON.stringify({ reasons, requestedAt: now.toISOString() }), + MessageGroupId: 'recompute', + MessageDeduplicationId: recomputeDeduplicationId(now.getTime()), + }) + ); + logger.info('Queued an event-triggered recompute', { reasons }); + return true; + } catch (error) { + logger.warn(`Recompute signal failed; the scheduled run still covers it: ${error}`); + return false; + } +} + async function processEvent( event: any, scope: CollectionScope, diff --git a/lambdas/incremental-collector/src/recompute-signal.test.ts b/lambdas/incremental-collector/src/recompute-signal.test.ts new file mode 100644 index 00000000..613ed462 --- /dev/null +++ b/lambdas/incremental-collector/src/recompute-signal.test.ts @@ -0,0 +1,51 @@ +import { signalRecompute } from '../index'; + +const environment = { + RECOMPUTE_QUEUE_URL: 'https://sqs.us-west-2.amazonaws.com/111122223333/khalifa-recompute.fifo', +}; +const role = { id: 'arn:aws:iam::111122223333:role/app', label: 'IamRole', properties: {} }; +const bucket = { id: 'arn:aws:s3:::data', label: 'S3Bucket', properties: {} }; +const now = new Date('2026-10-27T12:03:00Z'); + +describe('signalRecompute', () => { + it('queues one message for an IAM-sensitive batch, with a five-minute dedup id', async () => { + const send = jest.fn().mockResolvedValue({}); + await expect( + signalRecompute( + { send } as never, + environment, + [role], + [{ from: 'r', to: 'ext', label: 'TRUSTS' }], + now + ) + ).resolves.toBe(true); + + expect(send).toHaveBeenCalledTimes(1); + const input = send.mock.calls[0][0].input; + expect(input.QueueUrl).toBe(environment.RECOMPUTE_QUEUE_URL); + expect(input.MessageGroupId).toBe('recompute'); + expect(input.MessageDeduplicationId).toMatch(/^recompute-\d+$/); + expect(JSON.parse(input.MessageBody)).toMatchObject({ + reasons: ['IamRole', 'TRUSTS'], + requestedAt: '2026-10-27T12:03:00.000Z', + }); + }); + + it('sends nothing for non-IAM batches or without a queue', async () => { + const send = jest.fn().mockResolvedValue({}); + await expect(signalRecompute({ send } as never, environment, [bucket], [], now)).resolves.toBe( + false + ); + await expect(signalRecompute({ send } as never, {}, [role], [], now)).resolves.toBe(false); + expect(send).not.toHaveBeenCalled(); + }); + + it('never fails collection when the signal cannot be sent', async () => { + const warn = jest.spyOn(console, 'warn').mockImplementation(() => undefined); + const send = jest.fn().mockRejectedValue(new Error('AccessDenied')); + await expect(signalRecompute({ send } as never, environment, [role], [], now)).resolves.toBe( + false + ); + warn.mockRestore(); + }); +}); diff --git a/lambdas/recompute-worker/index.test.ts b/lambdas/recompute-worker/index.test.ts new file mode 100644 index 00000000..d45af514 --- /dev/null +++ b/lambdas/recompute-worker/index.test.ts @@ -0,0 +1,125 @@ +import { recomputeMinIntervalSeconds, runRecomputeWorker } from './index'; + +const environment = { + AWS_REGION: 'us-west-2', + RECOMPUTE_QUEUE_URL: 'https://sqs.us-west-2.amazonaws.com/111122223333/khalifa-recompute.fifo', + NEPTUNE_ENDPOINT: 'cluster:8182', +}; + +// Serves scripted receive batches, then aborts the loop so the test ends. +function sqsWithBatches( + batches: { Messages?: { ReceiptHandle?: string }[] }[], + controller: AbortController +): any { + const received: number[] = []; + return { + received, + send: jest.fn( + async (command: { constructor: { name: string }; input: Record }) => { + if (command.constructor.name === 'ReceiveMessageCommand') { + const batch = batches.shift(); + if (!batch) { + controller.abort(); + return { Messages: [] }; + } + received.push(batch.Messages?.length ?? 0); + return batch; + } + return {}; + } + ), + }; +} + +const message = (id: string) => ({ ReceiptHandle: `handle-${id}` }); + +describe('recomputeMinIntervalSeconds', () => { + it('defaults to fifteen minutes and validates bounds', () => { + expect(recomputeMinIntervalSeconds({})).toBe(900); + expect(recomputeMinIntervalSeconds({ RECOMPUTE_MIN_INTERVAL_SECONDS: '300' })).toBe(300); + expect(recomputeMinIntervalSeconds({ RECOMPUTE_MIN_INTERVAL_SECONDS: '0' })).toBe(0); + expect(() => recomputeMinIntervalSeconds({ RECOMPUTE_MIN_INTERVAL_SECONDS: 'x' })).toThrow( + 'RECOMPUTE_MIN_INTERVAL_SECONDS must be an integer between 0 and 86400' + ); + expect(() => recomputeMinIntervalSeconds({ RECOMPUTE_MIN_INTERVAL_SECONDS: '90000' })).toThrow( + 'between 0 and 86400' + ); + }); +}); + +describe('runRecomputeWorker', () => { + it('evaluates once per interval, deleting covered messages', async () => { + const controller = new AbortController(); + const sqs = sqsWithBatches( + [{ Messages: [message('1'), message('2')] }, { Messages: [message('3')] }], + controller + ); + const clock = Date.parse('2026-10-27T12:00:00Z'); + const evaluate = jest.fn().mockResolvedValue({}); + + const result = await runRecomputeWorker(environment, { + sqsClient: sqs, + evaluate: evaluate as never, + now: () => new Date(clock), + signal: controller.signal, + }); + + expect(result).toEqual({ messages: 3, evaluations: 1, skipped: 1 }); + expect(evaluate).toHaveBeenCalledWith({}, environment); + const deletes = sqs.send.mock.calls.filter( + ([command]: [{ constructor: { name: string } }]) => + command.constructor.name === 'DeleteMessageCommand' + ); + expect(deletes).toHaveLength(3); + }); + + it('evaluates again once the interval has elapsed', async () => { + const controller = new AbortController(); + const sqs = sqsWithBatches( + [{ Messages: [message('1')] }, { Messages: [message('2')] }], + controller + ); + let clock = Date.parse('2026-10-27T12:00:00Z'); + const evaluate = jest.fn().mockImplementation(async () => { + clock += 901_000; // next batch lands outside the 900s interval + }); + + const result = await runRecomputeWorker(environment, { + sqsClient: sqs, + evaluate: evaluate as never, + now: () => new Date(clock), + signal: controller.signal, + }); + + expect(result.evaluations).toBe(2); + expect(result.skipped).toBe(0); + }); + + it('leaves messages for redelivery when evaluation fails', async () => { + const controller = new AbortController(); + const sqs = sqsWithBatches([{ Messages: [message('1')] }], controller); + const error = jest.spyOn(console, 'error').mockImplementation(() => undefined); + const evaluate = jest.fn().mockRejectedValue(new Error('Neptune unavailable')); + const current = Date.now(); + const now = () => new Date(current); + + const result = await runRecomputeWorker(environment, { + sqsClient: sqs, + evaluate: evaluate as never, + now, + signal: controller.signal, + }); + + error.mockRestore(); + expect(result).toEqual({ messages: 1, evaluations: 0, skipped: 0 }); + const deletes = sqs.send.mock.calls.filter( + ([command]: [{ constructor: { name: string } }]) => + command.constructor.name === 'DeleteMessageCommand' + ); + expect(deletes).toHaveLength(0); + }); + + it('requires the queue url', async () => { + await expect(runRecomputeWorker({}, {})).rejects.toThrow('RECOMPUTE_QUEUE_URL is required'); + }); +}); diff --git a/lambdas/recompute-worker/index.ts b/lambdas/recompute-worker/index.ts new file mode 100644 index 00000000..b1760a46 --- /dev/null +++ b/lambdas/recompute-worker/index.ts @@ -0,0 +1,104 @@ +import { DeleteMessageCommand, ReceiveMessageCommand, SQSClient } from '@aws-sdk/client-sqs'; +import { runPolicyEvaluation } from '../policy-evaluator'; +import { Logger } from '../shared/types'; + +// Event-triggered recompute: the incremental collector signals when an IAM +// change could have moved an escalation, and this worker drains that queue by +// re-running the policy evaluator — so a new admin grant or trust produces its +// escalation paths within one incremental cycle instead of at the evaluator's +// scheduled hour. Two bounds keep the heavy job from running constantly: the +// producer's FIFO deduplication id collapses bursts to one message per +// five-minute window, and the worker itself refuses to run more often than +// RECOMPUTE_MIN_INTERVAL_SECONDS (default 900). Messages received inside the +// interval are deleted as covered by the run that just happened; messages +// whose evaluation failed are left for SQS redelivery (and the queue's DLQ). +const logger = new Logger('recompute-worker'); +export const DEFAULT_RECOMPUTE_MIN_INTERVAL_SECONDS = 900; +const MAX_MESSAGES = 10; + +export interface RecomputeWorkerResult { + messages: number; + evaluations: number; + skipped: number; +} + +export function recomputeMinIntervalSeconds( + environment: NodeJS.ProcessEnv, + name = 'RECOMPUTE_MIN_INTERVAL_SECONDS' +): number { + const raw = environment[name] ?? String(DEFAULT_RECOMPUTE_MIN_INTERVAL_SECONDS); + const parsed = + typeof raw === 'number' ? raw : /^\d+$/.test(String(raw).trim()) ? Number(raw) : NaN; + if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > 86_400) { + throw new Error(`${name} must be an integer between 0 and 86400`); + } + return parsed; +} + +export interface RecomputeWorkerDependencies { + sqsClient?: SQSClient; + evaluate?: typeof runPolicyEvaluation; + now?: () => Date; + signal?: AbortSignal; +} + +export async function runRecomputeWorker( + environment: NodeJS.ProcessEnv = process.env, + dependencies: RecomputeWorkerDependencies = {} +): Promise { + const queueUrl = environment.RECOMPUTE_QUEUE_URL?.trim(); + if (!queueUrl) throw new Error('RECOMPUTE_QUEUE_URL is required'); + const minIntervalMs = recomputeMinIntervalSeconds(environment) * 1000; + const sqsClient = + dependencies.sqsClient || new SQSClient({ region: environment.AWS_REGION || 'us-east-1' }); + const evaluate = dependencies.evaluate || runPolicyEvaluation; + const now = dependencies.now || (() => new Date()); + const signal = dependencies.signal; + + const result: RecomputeWorkerResult = { messages: 0, evaluations: 0, skipped: 0 }; + let lastRunAt: number | undefined; + + while (!signal?.aborted) { + const response = await sqsClient.send( + new ReceiveMessageCommand({ + QueueUrl: queueUrl, + MaxNumberOfMessages: MAX_MESSAGES, + WaitTimeSeconds: 20, + }), + signal ? { abortSignal: signal } : undefined + ); + const messages = response.Messages || []; + if (messages.length === 0) continue; + result.messages += messages.length; + + const nowMs = now().getTime(); + let evaluated = true; + if (lastRunAt !== undefined && nowMs - lastRunAt < minIntervalMs) { + result.skipped += messages.length; + logger.info('Recompute skipped inside the minimum interval', { + nextAllowedInSeconds: Math.ceil((minIntervalMs - (nowMs - lastRunAt)) / 1000), + }); + } else { + try { + await evaluate({}, environment); + lastRunAt = nowMs; + result.evaluations += 1; + logger.info('Event-triggered policy evaluation complete'); + } catch (error) { + evaluated = false; + logger.error( + `Event-triggered policy evaluation failed; leaving messages for redelivery: ${error}` + ); + } + } + + if (!evaluated) continue; + for (const message of messages) { + if (!message.ReceiptHandle) continue; + await sqsClient.send( + new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: message.ReceiptHandle }) + ); + } + } + return result; +} diff --git a/lambdas/recompute-worker/package.json b/lambdas/recompute-worker/package.json new file mode 100644 index 00000000..8a543853 --- /dev/null +++ b/lambdas/recompute-worker/package.json @@ -0,0 +1,33 @@ +{ + "name": "recompute-worker", + "version": "1.0.0", + "private": true, + "description": "Event-triggered policy evaluation: drains the recompute queue with a rate bound", + "main": "dist/index.js", + "scripts": { + "build": "esbuild index.ts --bundle --platform=node --format=cjs --target=node24 --outfile=dist/index.js --sourcemap --sources-content=false", + "typecheck": "tsc --noEmit", + "lint": "eslint . --ext .ts", + "format": "prettier --write \"**/*.ts\"", + "format:check": "prettier --check \"**/*.ts\"", + "test": "jest --passWithNoTests", + "test:watch": "jest --watch" + }, + "dependencies": { + "@aws-sdk/client-sqs": "^3.400.0", + "@khalifa/risk-engine": "*" + }, + "devDependencies": { + "@types/jest": "^29.5.0", + "@types/node": "^20.10.0", + "jest": "^29.7.0", + "ts-jest": "^29.4.11", + "typescript": "^5.3.2", + "eslint": "^8.56.0", + "prettier": "^3.2.5", + "@typescript-eslint/eslint-plugin": "^7.0.0", + "@typescript-eslint/parser": "^7.0.0", + "eslint-config-prettier": "^9.1.0", + "eslint-plugin-prettier": "^5.1.3" + } +} diff --git a/lambdas/recompute-worker/tsconfig.json b/lambdas/recompute-worker/tsconfig.json new file mode 100644 index 00000000..eb9e8c88 --- /dev/null +++ b/lambdas/recompute-worker/tsconfig.json @@ -0,0 +1,20 @@ +{ + "compilerOptions": { + "target": "ES2020", + "module": "commonjs", + "lib": ["ES2020"], + "outDir": "./dist", + "rootDir": "..", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "resolveJsonModule": true, + "declaration": true, + "declarationMap": true, + "sourceMap": true, + "moduleResolution": "node" + }, + "include": ["../recompute-worker/*.ts", "../policy-evaluator/**/*", "../shared/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/lambdas/runtime/index.test.ts b/lambdas/runtime/index.test.ts index 19e38daa..fc6d3722 100644 --- a/lambdas/runtime/index.test.ts +++ b/lambdas/runtime/index.test.ts @@ -28,6 +28,12 @@ const analyzerEnvironment = { ACCESS_ANALYZER_TABLE: 'AccessAnalyzerCache', }; +const recomputeEnvironment = { + AWS_REGION: 'us-west-2', + NEPTUNE_ENDPOINT: 'cluster:8182', + RECOMPUTE_QUEUE_URL: 'https://sqs.us-west-2.amazonaws.com/111122223333/khalifa-recompute.fifo', +}; + const threatIntelEnvironment = { AWS_REGION: 'us-west-2', THREAT_INDICATORS_TABLE: 'khalifa-threat-indicators', @@ -86,6 +92,16 @@ describe('Kubernetes runtime', () => { ).toThrow('GRAPH_WRITER_TOKEN is required'); expect(() => validateRuntime('flow-log-analyzer', flowLogEnvironment)).not.toThrow(); expect(() => validateRuntime('threat-intel-import', threatIntelEnvironment)).not.toThrow(); + expect(() => validateRuntime('recompute-worker', recomputeEnvironment)).not.toThrow(); + expect(() => + validateRuntime('recompute-worker', { AWS_REGION: 'us-west-2', NEPTUNE_ENDPOINT: 'x' }) + ).toThrow('RECOMPUTE_QUEUE_URL is required'); + expect(() => + validateRuntime('recompute-worker', { + ...recomputeEnvironment, + RECOMPUTE_MIN_INTERVAL_SECONDS: 'nope', + }) + ).toThrow('RECOMPUTE_MIN_INTERVAL_SECONDS must be an integer between 0 and 86400'); expect(() => validateRuntime('threat-intel-import', { AWS_REGION: 'us-west-2', @@ -207,6 +223,23 @@ describe('Kubernetes runtime', () => { ).rejects.toThrow('NoSuchKey'); }); + it('starts the recompute worker with the signal and reports its summary', async () => { + const worker = jest.fn().mockResolvedValue({ messages: 2, evaluations: 1, skipped: 1 }); + await expect( + runCommand('recompute-worker', recomputeEnvironment, new AbortController().signal, { + runRecomputeWorker: worker, + }) + ).resolves.toBeUndefined(); + expect(worker).toHaveBeenCalledWith(recomputeEnvironment, { signal: expect.any(AbortSignal) }); + + const failing = jest.fn().mockRejectedValue(new Error('boom')); + await expect( + runCommand('recompute-worker', recomputeEnvironment, new AbortController().signal, { + runRecomputeWorker: failing, + }) + ).rejects.toThrow('boom'); + }); + it('runs the DSPM scanner once with the HTTP graph writer', async () => { const runDspm = jest.fn().mockResolvedValue({ bucketsScanned: 2, diff --git a/lambdas/runtime/index.ts b/lambdas/runtime/index.ts index a60f8bc8..77709a6b 100644 --- a/lambdas/runtime/index.ts +++ b/lambdas/runtime/index.ts @@ -29,6 +29,7 @@ import { } from '../flow-log-analyzer'; import { validateObservationDays } from '../flow-log-analyzer/src/queries'; import { importThreatIntel, parseS3Uri, validateTtlDays } from '../threat-intel-import'; +import { recomputeMinIntervalSeconds, runRecomputeWorker } from '../recompute-worker'; export { createHttpGraphWriter } from '../shared/graph-writer-client'; @@ -41,6 +42,7 @@ export type RuntimeCommand = | 'cloudtrail-analyzer' | 'flow-log-analyzer' | 'threat-intel-import' + | 'recompute-worker' | 'dspm-scanner' | 'policy-evaluator' | 'least-privilege-sweep'; @@ -55,6 +57,7 @@ interface RuntimeDependencies { analyzeCloudTrail?: typeof analyzeCloudTrail; analyzeFlowLogs?: typeof analyzeFlowLogs; importThreatIntel?: typeof importThreatIntel; + runRecomputeWorker?: typeof runRecomputeWorker; runDspm?: typeof runDspmScan; runPolicyEvaluation?: typeof runPolicyEvaluation; runLeastPrivilegeSweep?: typeof runLeastPrivilegeSweep; @@ -73,6 +76,7 @@ const commands = new Set([ 'cloudtrail-analyzer', 'flow-log-analyzer', 'threat-intel-import', + 'recompute-worker', 'dspm-scanner', 'policy-evaluator', 'least-privilege-sweep', @@ -89,6 +93,7 @@ Commands: cloudtrail-analyzer Query org CloudTrail usage into AccessAnalyzerCache flow-log-analyzer Aggregate VPC flow logs into TrafficObservations threat-intel-import Import customer threat-intel lists into ThreatIndicators + recompute-worker Drain the recompute queue and re-run policy evaluation dspm-scanner Scan S3 objects and Lambda env vars for secrets policy-evaluator Resolve effective permissions and escalation paths once least-privilege-sweep Rightsize over-permissioned principals once`; @@ -214,6 +219,13 @@ export function validateRuntime(command: RuntimeCommand, environment: NodeJS.Pro validateObservationDays(environment.FLOW_LOG_DAYS ?? '7', 'FLOW_LOG_DAYS'); return; } + if (command === 'recompute-worker') { + required(environment, 'AWS_REGION'); + required(environment, 'RECOMPUTE_QUEUE_URL'); + required(environment, 'NEPTUNE_ENDPOINT'); + recomputeMinIntervalSeconds(environment); + return; + } if (command === 'threat-intel-import') { required(environment, 'AWS_REGION'); required(environment, 'THREAT_INDICATORS_TABLE'); @@ -680,6 +692,11 @@ export async function runCommand( environment ); logger.info('Threat-intel import complete', summary); + } else if (command === 'recompute-worker') { + const summary = await (dependencies.runRecomputeWorker || runRecomputeWorker)(environment, { + signal, + }); + logger.info('Recompute worker stopped', summary); } else if (command === 'dspm-scanner') { const summary = await (dependencies.runDspm || runDspmScan)({}, {}, environment); logger.info('DSPM scan complete', summary); diff --git a/lambdas/shared/recompute.test.ts b/lambdas/shared/recompute.test.ts new file mode 100644 index 00000000..1d47473c --- /dev/null +++ b/lambdas/shared/recompute.test.ts @@ -0,0 +1,40 @@ +import { recomputeDeduplicationId, recomputeReasons, RECOMPUTE_DEDUP_WINDOW_MS } from './recompute'; + +describe('recomputeReasons', () => { + it('names the IAM node labels and edge labels that make a batch stale', () => { + const reasons = recomputeReasons( + [ + { id: 'role', label: 'IamRole', properties: {} }, + { id: 'bucket', label: 'S3Bucket', properties: {} }, + { id: 'policy', label: 'IamPolicyDocument', properties: {} }, + ], + [ + { from: 'role', to: 'ext', label: 'TRUSTS' }, + { from: 'img', to: 'repo', label: 'BELONGS_TO' }, + ] + ); + expect(reasons).toEqual(['IamPolicyDocument', 'IamRole', 'TRUSTS']); + }); + + it('is empty for collection that cannot move an escalation', () => { + expect( + recomputeReasons( + [{ id: 'bucket', label: 'S3Bucket', properties: {} }], + [{ from: 'img', to: 'repo', label: 'BELONGS_TO' }] + ) + ).toEqual([]); + }); +}); + +describe('recomputeDeduplicationId', () => { + it('buckets ids into five-minute windows', () => { + const start = 1_700_000_100_000; // on a five-minute boundary + expect(recomputeDeduplicationId(start)).toBe(recomputeDeduplicationId(start + 1_000)); + expect(recomputeDeduplicationId(start + RECOMPUTE_DEDUP_WINDOW_MS - 1)).toBe( + recomputeDeduplicationId(start) + ); + expect(recomputeDeduplicationId(start + RECOMPUTE_DEDUP_WINDOW_MS)).not.toBe( + recomputeDeduplicationId(start) + ); + }); +}); diff --git a/lambdas/shared/recompute.ts b/lambdas/shared/recompute.ts new file mode 100644 index 00000000..9e4037ee --- /dev/null +++ b/lambdas/shared/recompute.ts @@ -0,0 +1,42 @@ +import type { GraphEdge, GraphNode } from './types'; + +// Event-triggered recompute: which graph changes make yesterday's evaluation +// stale? IAM is the whole CIEM surface — a new grant, role, key or trust can +// create an escalation the policy evaluator only writes on its schedule — so +// any node or edge in that family marks a batch worth recomputing for. +export const RECOMPUTE_NODE_LABELS = new Set([ + 'IamGroup', + 'IamPolicy', + 'IamPolicyDocument', + 'IamPolicyStatement', + 'IamRole', + 'IamUser', + 'AccessKey', +]); + +export const RECOMPUTE_EDGE_LABELS = new Set([ + 'ATTACHED_TO', + 'HAS_ACCESS_KEY', + 'HAS_IAM_ROLE', + 'MEMBER_OF', + 'TRUSTS', +]); + +export function recomputeReasons(nodes: GraphNode[], edges: GraphEdge[]): string[] { + const reasons = new Set(); + for (const node of nodes) { + if (RECOMPUTE_NODE_LABELS.has(node.label)) reasons.add(node.label); + } + for (const edge of edges) { + if (RECOMPUTE_EDGE_LABELS.has(edge.label)) reasons.add(edge.label); + } + return [...reasons].sort(); +} + +// FIFO dedup window: SQS only keeps a deduplication id for five minutes, so a +// burst of IAM events in one window collapses to a single recompute message. +export const RECOMPUTE_DEDUP_WINDOW_MS = 5 * 60 * 1000; + +export function recomputeDeduplicationId(nowMs: number): string { + return `recompute-${Math.floor(nowMs / RECOMPUTE_DEDUP_WINDOW_MS)}`; +} diff --git a/package-lock.json b/package-lock.json index 8ed03fb4..1d514bc0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -418,6 +418,36 @@ "undici-types": "~6.21.0" } }, + "lambdas/recompute-worker": { + "version": "1.0.0", + "dependencies": { + "@aws-sdk/client-sqs": "^3.400.0", + "@khalifa/risk-engine": "*" + }, + "devDependencies": { + "@types/jest": "^29.5.0", + "@types/node": "^20.10.0", + "@typescript-eslint/eslint-plugin": "^7.0.0", + "@typescript-eslint/parser": "^7.0.0", + "eslint": "^8.56.0", + "eslint-config-prettier": "^9.1.0", + "eslint-plugin-prettier": "^5.1.3", + "jest": "^29.7.0", + "prettier": "^3.2.5", + "ts-jest": "^29.4.11", + "typescript": "^5.3.2" + } + }, + "lambdas/recompute-worker/node_modules/@types/node": { + "version": "20.19.43", + "resolved": "https://registry.npmjs.org/@types/node/-/node-20.19.43.tgz", + "integrity": "sha512-6oYBAi5ikg4Pl+kGsoYtawUMBT2zZMCvPNF7pVLnHZfd1zf38DRiWn/gT01RYCdUqkv7Fhr+C9ot4/tb+2sVvA==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~6.21.0" + } + }, "lambdas/runtime": { "name": "khalifa-runtime", "version": "1.0.0", @@ -11995,6 +12025,10 @@ "node": ">=8.10.0" } }, + "node_modules/recompute-worker": { + "resolved": "lambdas/recompute-worker", + "link": true + }, "node_modules/redent": { "version": "3.0.0", "resolved": "https://registry.npmjs.org/redent/-/redent-3.0.0.tgz",