From d72b196a94b615b0296045d4571bb7089132becc Mon Sep 17 00:00:00 2001 From: Adron Hall Date: Wed, 21 Oct 2026 20:13:12 -0700 Subject: [PATCH] Add Audit Event exports (POST/GET /v1/audit-exports) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The audit log can be exported to a file for compliance evidence (API Reference → Audit → Exporting), in the shape the console needs (#56): - POST /v1/audit-exports: the same filters and caller scope as GET /v1/audit-events, plus format ndjson|csv; Idempotency-Key required; 202 with the export. 20 per caller per day. - GET /v1/audit-exports/{id}: status, and once ready a download link signed for 15 minutes. Requester and audit.view only. Kept 7 days. - A jobs worker (audit-exports, every minute; every 15 s in the devserver) pages through the events into a temp file and uploads it through a new ExportStore: a private S3 bucket with a 7-day lifecycle rule in AWS, .dev/audit-exports locally, in memory in tests. - audit.exported is written when the file is ready, with the requester as actor, in the export's own scope. - audit.go's scope and filters are shared by the list and the worker. - Migration 0007_audit_exports (after 0005 and 0006, keeping user.expired in the action check). Terraform: the exports bucket, env, IAM, and schedule. Retention needed no code: the hot window and archival already enforce it; the docs now say so. Closes #33 Closes #56 Co-Authored-By: Claude Opus 5.5 --- cmd/devserver/main.go | 12 +- docs/api-reference/audit.md | 39 ++- docs/changelog.md | 12 + docs/domain-model/database-schema.md | 18 ++ docs/domain-model/orders-and-audit.md | 1 + docs/non-functional-requirements.md | 1 + docs/openapi.yaml | 83 +++++- docs/openapi/index.md | 2 +- docs/roadmap.md | 2 +- infra/README.md | 4 + infra/terraform/prod/main.tf | 44 +++- internal/api/gen/gen.go | 231 +++++++++++++++++ internal/core/audit.go | 127 ++++++---- internal/core/audit_exports.go | 347 ++++++++++++++++++++++++++ internal/core/audit_exports_test.go | 119 +++++++++ internal/core/server.go | 1 + internal/core/system.go | 1 + internal/ids/ids.go | 1 + internal/jobs/jobs.go | 6 + internal/mcp/tools.json | 201 ++++++++++++++- internal/platform/config/config.go | 5 + internal/platform/wire/deps.go | 10 +- internal/platform/wire/exports.go | 78 ++++++ internal/testenv/testenv.go | 44 +++- migrations/0007_audit_exports.sql | 54 ++++ plan-remaining.md | 1 - 26 files changed, 1379 insertions(+), 65 deletions(-) create mode 100644 internal/core/audit_exports.go create mode 100644 internal/core/audit_exports_test.go create mode 100644 internal/platform/wire/exports.go create mode 100644 migrations/0007_audit_exports.sql diff --git a/cmd/devserver/main.go b/cmd/devserver/main.go index 2dc1f3c..df46dc7 100644 --- a/cmd/devserver/main.go +++ b/cmd/devserver/main.go @@ -56,20 +56,22 @@ func main() { Stripe: stripe, Core: server, Log: base.Log}) } - // The 5-minute entitlement sweep, in-process. + // The 5-minute entitlement sweep and the audit-export worker, in-process. runner := jobs.New(base) - go func() { - t := time.NewTicker(5 * time.Minute) + every := func(d time.Duration, job string) { + t := time.NewTicker(d) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: - _ = runner.Run(ctx, "entitlement-sweep") + _ = runner.Run(ctx, job) } } - }() + } + go every(5*time.Minute, "entitlement-sweep") + go every(15*time.Second, "audit-exports") srv := &http.Server{Addr: base.Config.ListenAddr, Handler: mux, ReadHeaderTimeout: 10 * time.Second} go func() { diff --git a/docs/api-reference/audit.md b/docs/api-reference/audit.md index 5afc8f0..974fd9c 100644 --- a/docs/api-reference/audit.md +++ b/docs/api-reference/audit.md @@ -22,8 +22,10 @@ See [Domain Model → Audit Event](../../domain-model/orders-and-audit/#audit-ev |---|---|---|---| | `GET` | `/v1/audit-events` | `audit.view` (platform or app-confined); or self via `target_user_id=me`; or a Tenant owner via `tenant_id` | Query the audit log. | | `GET` | `/v1/audit-events/{id}` | same as above, for that event | Fetch one event. | +| `POST` | `/v1/audit-exports` | same as `GET /v1/audit-events`, for the filters sent | Export matching events to a file. | +| `GET` | `/v1/audit-exports/{id}` | the export's requester, or `audit.view` | Check an export, and get its download link. | -There's deliberately no write endpoint. Audit Events are produced only as a side effect of other writes, never created directly, and never edited or deleted through the API. +There's deliberately no write endpoint for Audit Events. They're produced only as a side effect of other writes, never created directly, and never edited or deleted through the API. An export is a read: it copies events into a file and changes none of them. ## `GET /v1/audit-events` @@ -79,6 +81,40 @@ GET /v1/audit-events?target_user_id=usr_01JAG3Z9X8QS3F6K2M4N5P6R7S&since=2026-09 Returns one event, same shape. `404 audit_event_not_found` if it doesn't exist, isn't visible to the caller, or has been archived. +## Exporting + +For compliance evidence (a SOC 2 audit, a customer's request for their records) or offline analysis, export the log to a file instead of paging through it. + +```json +// POST /v1/audit-exports (Idempotency-Key required) +{ "format": "csv", "tenant_id": "tnt_01JAG1SYSTEM00000000000000", "since": "2026-01-01T00:00:00Z" } +``` +```json +// Response — 202 +{ + "id": "aex_01JAH2K4Q9R0S1T2U3V4W5X6Y7", + "status": "pending", + "format": "csv", + "filters": { "tenant_id": "tnt_01JAG1SYSTEM00000000000000", "since": "2026-01-01T00:00:00Z" }, + "row_count": null, + "download_url": null, + "download_url_expires_at": null, + "error": null, + "created_at": "2026-10-21T18:00:00Z", + "completed_at": null, + "expires_at": "2026-10-28T18:00:00Z", + "test_mode": false +} +``` + +- **Filters and scope** are exactly `GET /v1/audit-events`'s, as body fields: `action` is an array (up to 100), `target_user_id` may be `me`, and the caller is confined to the same scope as on the list. An app-confined key exports only its own Application's events, a User without `audit.view` only their own trail (`target_user_id: "me"`) or a Tenant they own. Hot window only, like the list. +- **`format`** is `ndjson` (default), one Audit Event object per line, or `csv`, with the columns `id, timestamp, action, actor_type, actor_id, target_type, target_id, target_user_id, application_id, organization_id, tenant_id, request_id, before, after` (`before`/`after` as JSON text). Rows are oldest first. +- **Asynchronous.** The request returns `202` at once, and a job writes the file, usually within a minute. Poll `GET /v1/audit-exports/{id}` until `status` is `ready` (or `failed`, with `error` saying why: `too_many_rows` over 1,000,000 events, so narrow the filters). +- **Downloading.** Once `ready`, `download_url` is a signed link valid for 15 minutes (`download_url_expires_at`). Fetch the export again for a fresh one. The file and the record are deleted 7 days after the request (`expires_at`). +- **Who can see it:** whoever requested it, and `audit.view`. Anyone else gets `404 audit_export_not_found`. +- **It's audited.** When the file is ready, `audit.exported` is written with the requester as actor, the filters, the format, and the row count, in the export's own scope, so it appears in the same trail it exported from. +- Limited to 20 per caller per day. + ## Retention Audit Events are never deleted or edited through the API, including after the record they describe is hard-deleted. For example, a user's right-to-erasure request removes their Profile/Settings, but the audit trail of *that deletion itself* persists, with snapshot values redacted. See [Non-Functional Requirements → Data retention](../../non-functional-requirements/#data-retention). @@ -88,4 +124,5 @@ Audit Events are never deleted or edited through the API, including after the re | Code | Status | When | |---|---|---| | `audit_event_not_found` | 404 | See above. | +| `audit_export_not_found` | 404 | The export doesn't exist, has expired, or wasn't requested by the caller (and they lack `audit.view`). | | `forbidden` | 403 | No `audit.view`, and the query isn't scoped to `me` or an owned Tenant. | diff --git a/docs/changelog.md b/docs/changelog.md index 0a5871a..6f08ba7 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -12,6 +12,18 @@ What's changed in this specification over time. Every decision is written into t --- +## 2026-10-21 — Audit Event exports + +The audit log can be exported to a file. See [API Reference → Audit → Exporting](../api-reference/audit/#exporting). `openapi.yaml` is now at `1.0.0-draft.4`, with 120 operations. + +- **`POST /v1/audit-exports`** takes the same filters as `GET /v1/audit-events`, as body fields, plus `format` (`ndjson` or `csv`). It confines the caller to the same scope as the list, requires `Idempotency-Key`, and returns `202`. A job writes the file, usually within a minute. Limited to 20 per caller per day. +- **`GET /v1/audit-exports/{id}`** shows the status, and once it's `ready`, a download link signed for 15 minutes. Only the requester and `audit.view` can see an export. Files and records are deleted after 7 days. +- New audit action `audit.exported` (target type `audit_export`), written when the file is ready, with the requester as actor. +- **Retention** needed no new code. It's already enforced: the one-year hot window, and the nightly archival of older events, shape-only, to indefinite cold storage ([Audit log lifecycle](../non-functional-requirements/#audit-log-lifecycle)). +- Migration `0007_audit_exports`. Infrastructure: a private exports bucket with a 7-day lifecycle rule, and an `audit-exports` schedule every minute. + +--- + ## 2026-10-20 — Test-mode Users are time-limited Test mode already never counted toward billing (decision #36, confirmed on 2026-10-18). Now every **test-mode User expires**, so a test account can't stand in for a free production account. See [Auth → Test-mode Users → Lifetime](../api-reference/auth/#lifetime). diff --git a/docs/domain-model/database-schema.md b/docs/domain-model/database-schema.md index f826569..7e5246d 100644 --- a/docs/domain-model/database-schema.md +++ b/docs/domain-model/database-schema.md @@ -438,6 +438,24 @@ No `updated_at`, no soft-delete column — this table is append-only by design ( **Enforcement, not just storage:** the `restrict_destructive` check happens in the same request-handling layer as the permission check — both read from the authenticated key's row, both must pass. The classification of which operations count as destructive lives in exactly one place in the implementation (a lookup by HTTP method + the specific Entitlement/Role/Organization-member/Tenant-tier mutations named in [API Keys → Agent keys & `restrict_destructive`](../../api-reference/api-keys/#agent-keys--restrict_destructive)), shared by this check and by the MCP server's `destructiveHint` tool-annotation generation — not reimplemented twice and left to drift. +## audit_exports + +| Column | Type | Constraint | +|---|---|---| +| `id` | `text` | PK, `aex_…` | +| `requested_by_type`, `requested_by_id` | `text` | not null — `user` or `api_key`, and its id | +| `scope` | `jsonb` | not null — the scope the requester was confined to (`application_id`, `target_user_id`, or `tenant_id`), or `{}` for `audit.view` | +| `filters` | `jsonb` | not null — as requested, `me` resolved | +| `format` | `text` | `ndjson` or `csv` | +| `status` | `text` | `pending`, `running`, `ready`, or `failed` | +| `row_count`, `object_key` | `integer`, `text` | set exactly when `ready` | +| `error` | `text` | set exactly when `failed` | +| `test_mode` | `boolean` | mode policy, like every other test-mode table | +| `created_at`, `started_at`, `completed_at` | `timestamptz` | | +| `expires_at` | `timestamptz` | not null — `created_at` + 7 days; the nightly cleanup deletes the row, and an S3 lifecycle rule the file | + +**Index:** `(created_at) where status in ('pending','running')` for the export worker, which claims one at a time with `for update skip locked` and picks up a `running` export again if its worker hasn't finished within 20 minutes. + ## webhook_subscriptions, webhook_events, webhook_deliveries ```sql diff --git a/docs/domain-model/orders-and-audit.md b/docs/domain-model/orders-and-audit.md index baa5311..817980e 100644 --- a/docs/domain-model/orders-and-audit.md +++ b/docs/domain-model/orders-and-audit.md @@ -123,6 +123,7 @@ Every value `action` can take. It's mirrored exactly by the `AuditAction` enum i | `webhook.created` / `webhook.updated` / `webhook.deleted` / `webhook.secret_rotated` | [Webhook](../../api-reference/webhooks/) subscription lifecycle | `webhook` | | `push_credential.created` / `push_credential.updated` / `push_credential.deleted` | An Application's [push credential](../../api-reference/notifications/#push-credentials) is configured, changed, or removed. The platform marking one `failed` (or a change returning it to `active`) is `push_credential.updated` with actor `system`. Key material never appears in `before`/`after` | `push_credential` | | `device.deleted` | `users.manage` deletes another User's [Device](../../api-reference/notifications/#delete-v1usersiddevicesdeviceid) — **admin-only** | `device` | +| `audit.exported` | An [Audit Event export](../../api-reference/audit/#exporting) is written; actor is whoever requested it, with the filters, format, and row count in `after` | `audit_export` | This list is the authoritative source for `action` values. If an endpoint's page describes a write that isn't represented here, that's a spec bug; file it the same way as any other inconsistency. diff --git a/docs/non-functional-requirements.md b/docs/non-functional-requirements.md index cd360c6..58d7d03 100644 --- a/docs/non-functional-requirements.md +++ b/docs/non-functional-requirements.md @@ -71,6 +71,7 @@ Per-API-key token bucket, returning `429` with a `Retry-After` header on exhaust | `POST /v1/auth/token/refresh` | 30 requests/minute per session | | `GET /v1/users/{id}/export` | 5 per user per day | | `POST /v1/organizations` | 10 per user per day | +| `POST /v1/audit-exports` | 20 per API Key or User per day | | `POST /v1/users/{id}/notifications` | 600 requests/minute per API Key, and 60 notifications per recipient User per Application per rolling hour (`429 rate_limited`, `details.scope: "recipient"`). Platform alerts don't count toward the per-recipient limit. | | `POST /v1/users/{id}/devices/{deviceId}/test` | 10 per device per hour | diff --git a/docs/openapi.yaml b/docs/openapi.yaml index cc5019b..05dccee 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: Substratal Apps — Platform API - version: "1.0.0-draft.3" + version: "1.0.0-draft.4" summary: >- User accounts, auth, roles, per-app entitlements (the on/off access switch), profiles, settings, organizations, tenancy, subscription billing, @@ -1958,6 +1958,51 @@ paths: schema: { $ref: "#/components/schemas/AuditEvent" } "404": { $ref: "#/components/responses/NotFound" } + /audit-exports: + post: + operationId: audit.createExport + tags: [Audit] + summary: Export Audit Events to a file (asynchronous) + description: >- + Exports every Audit Event matching the same filters as GET /v1/audit-events, in the caller's + scope there, as NDJSON or CSV. Hot window only. Returns 202 at once; a job writes the file, + usually within a minute. Poll GET /v1/audit-exports/{id} until status is ready, then download + from download_url. Exports are kept 7 days. Writes audit.exported when the file is ready. + x-substratal-phase: phase3 + parameters: + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/AuditExportRequest" } + responses: + "202": + description: Export queued + content: + application/json: + schema: { $ref: "#/components/schemas/AuditExport" } + "403": { $ref: "#/components/responses/Forbidden" } + "422": { $ref: "#/components/responses/Unprocessable" } + "429": { $ref: "#/components/responses/TooManyRequests" } + + /audit-exports/{id}: + parameters: + - { name: id, in: path, required: true, schema: { type: string } } + get: + operationId: audit.getExport + tags: [Audit] + summary: Fetch an export's status, and its download link once ready + description: "Visible to whoever requested it, and to audit.view. download_url is freshly signed on every call and valid for 15 minutes." + x-substratal-phase: phase3 + responses: + "200": + description: The export + content: + application/json: + schema: { $ref: "#/components/schemas/AuditExport" } + "404": { $ref: "#/components/responses/NotFound" } + # ───────────────────────────── Webhooks ───────────────────────────── /webhooks: @@ -3418,6 +3463,7 @@ components: - user.mfa_enabled - user.mfa_disabled - user.mfa_reset + - audit.exported - entitlement.granted - entitlement.disabled - entitlement.revoked @@ -3463,7 +3509,40 @@ components: AuditTargetType: type: string - enum: [user, entitlement, role, role_assignment, application, organization, subscription, tenant, tier_change_request, api_key, webhook, push_credential, device] + enum: [user, entitlement, role, role_assignment, application, organization, subscription, tenant, tier_change_request, api_key, webhook, push_credential, device, audit_export] + + AuditExportRequest: + type: object + description: "The same filters as GET /v1/audit-events, as body fields, plus format." + properties: + format: { type: string, enum: [ndjson, csv], default: ndjson } + target_user_id: { type: string, description: "A User id, or me." } + target_type: { $ref: "#/components/schemas/AuditTargetType" } + target_id: { type: string } + actor_id: { type: string } + actor_type: { type: string, enum: [user, api_key, system] } + action: { type: array, items: { $ref: "#/components/schemas/AuditAction" }, maxItems: 100 } + application_id: { type: string } + organization_id: { type: string } + tenant_id: { type: string } + since: { type: string, format: date-time } + until: { type: string, format: date-time } + + AuditExport: + type: object + properties: + id: { type: string } + status: { type: string, enum: [pending, running, ready, failed] } + format: { type: string, enum: [ndjson, csv] } + filters: { type: object, description: "The filters as requested, with target_user_id me resolved to the caller's id." } + row_count: { type: [integer, "null"], description: "Set once ready." } + download_url: { type: [string, "null"], format: uri, description: "Set once ready. Signed, valid for 15 minutes; fetch the export again for a fresh one." } + download_url_expires_at: { type: [string, "null"], format: date-time } + error: { type: [string, "null"], description: "Set when failed, e.g. too_many_rows." } + created_at: { type: string, format: date-time } + completed_at: { type: [string, "null"], format: date-time } + expires_at: { type: string, format: date-time, description: "When the file and this record are deleted: 7 days after the request." } + test_mode: { type: boolean } AuditEvent: type: object diff --git a/docs/openapi/index.md b/docs/openapi/index.md index f4a3992..d3a7b43 100644 --- a/docs/openapi/index.md +++ b/docs/openapi/index.md @@ -27,7 +27,7 @@ The complete Platform API contract as an [OpenAPI 3.1](https://spec.openapis.org | Base URL | `https://api.substratalapps.com/v1` | | Authentication | `Authorization: Bearer `: a user access token or an API Key (`bearerAuth`) | | Content type | `application/json; charset=utf-8` | -| Operations | 118, every one with a stable `operationId` | +| Operations | 120, every one with a stable `operationId` | | Schemas | 109 reusable components under `components.schemas` | The prose pages under [API Reference](../api-reference/) are authoritative. `openapi.yaml` is kept in sync with them by hand, and the implementation's contract tests validate every response against it, so a drift between the two fails CI rather than reaching a client. See [Conventions → Machine-readable](../api-reference/conventions/#machine-readable). diff --git a/docs/roadmap.md b/docs/roadmap.md index b6719ec..402bca2 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -46,7 +46,7 @@ Enough to replace "someone manually emails a link and flips a flag in a spreadsh ## Phase 3 -- Richer audit/compliance views — filtering, export, retention policy enforcement; see [Compliance & Data Protection](../compliance/) for what's driving this (SOC 2 readiness, GDPR data-subject requests). +- Richer audit/compliance views — filtering, export, retention policy enforcement; see [Compliance & Data Protection](../compliance/) for what's driving this (SOC 2 readiness, GDPR data-subject requests). **Built 2026-10-21:** filtering on [`GET /v1/audit-events`](../api-reference/audit/), [exports](../api-reference/audit/#exporting), and retention through the [Audit log lifecycle](../non-functional-requirements/#audit-log-lifecycle). - SCIM-style provisioning for larger customers who want their own IdP to push user lifecycle events into the hub automatically. - The **hosted sign-in page** and the authorization-code + PKCE flow ([Auth → Getting an app token](../api-reference/auth/#getting-an-app-token)). The API side is specified now; this is the UI. It's required before any third-party Application goes live. - Self-service Application registration and review queue for third-party developers — the marketplace phase. See [App developer/publisher model](../domain-model/applications/#the-review-lifecycle). diff --git a/infra/README.md b/infra/README.md index a3ce6cf..cb109d7 100644 --- a/infra/README.md +++ b/infra/README.md @@ -257,6 +257,10 @@ The delivery worker reserves 5 concurrent executions, so a burst of deliveries c Cost is negligible at Tier 0 volume (Glacier Deep Archive is priced for exactly this shape of rarely-read, kept-forever data) — this is about bounding Aurora's hot-table size and honoring the redaction-on-age guarantee, not about saving money on S3 itself. +### Audit exports + +[`POST /v1/audit-exports`](https://compositecode.github.io/substratalapps.com/api-reference/audit/#exporting) files go to a second private bucket, `substratal-prod-audit-exports-` (SSE-S3, public access blocked), with a lifecycle rule that deletes every object after 7 days, matching the export record's `expires_at`. The `audit-exports` job runs every minute on the jobs Lambda, pages through the matching events into a temporary file, and uploads it (`AUDIT_EXPORT_BUCKET`, `s3:PutObject`). The API Lambda signs the 15-minute download links, so it holds `s3:GetObject` on that bucket only. Locally, exports are written under `.dev/audit-exports` and the devserver runs the worker every 15 seconds. + ## Local development The RDS Data API choice that keeps Tier 0 NAT-free has a real cost: it doesn't have a clean open-source local emulator, so a developer can't just point the production data-access code at "Data API, but local" the way they could with a plain Postgres driver. diff --git a/infra/terraform/prod/main.tf b/infra/terraform/prod/main.tf index dbdb918..1119939 100644 --- a/infra/terraform/prod/main.tf +++ b/infra/terraform/prod/main.tf @@ -140,6 +140,38 @@ resource "aws_s3_bucket_versioning" "audit_archive" { versioning_configuration { status = "Enabled" } } +# Finished Audit Event exports (POST /v1/audit-exports). Private; the API +# hands out 15-minute presigned GETs, and objects are deleted after 7 days, +# when the export record is. +resource "aws_s3_bucket" "audit_exports" { + bucket = "${local.prefix}-audit-exports-${data.aws_caller_identity.me.account_id}" +} + +resource "aws_s3_bucket_public_access_block" "audit_exports" { + bucket = aws_s3_bucket.audit_exports.id + block_public_acls = true + block_public_policy = true + ignore_public_acls = true + restrict_public_buckets = true +} + +resource "aws_s3_bucket_server_side_encryption_configuration" "audit_exports" { + bucket = aws_s3_bucket.audit_exports.id + rule { + apply_server_side_encryption_by_default { sse_algorithm = "AES256" } + } +} + +resource "aws_s3_bucket_lifecycle_configuration" "audit_exports" { + bucket = aws_s3_bucket.audit_exports.id + rule { + id = "expire-exports" + status = "Enabled" + filter {} + expiration { days = 7 } + } +} + # ── Email (SES) ──────────────────────────────────────────────────────── resource "aws_sesv2_email_identity" "mail" { @@ -227,9 +259,13 @@ module "api" { environment = merge(local.common_env, { WEBHOOK_QUEUE_URL = aws_sqs_queue.webhooks.url CORS_ALLOWED_ORIGINS = var.cors_origins + AUDIT_EXPORT_BUCKET = aws_s3_bucket.audit_exports.bucket }) policy_statements = concat(local.data_api, local.secrets, local.sign, local.crypt, local.email, [{ Effect = "Allow", Action = ["sqs:SendMessage"], Resource = aws_sqs_queue.webhooks.arn + }, { + # Presigned download links are signed with this role. + Effect = "Allow", Action = ["s3:GetObject"], Resource = "${aws_s3_bucket.audit_exports.arn}/*" }]) } @@ -298,11 +334,14 @@ module "jobs" { artifact_key = local.artifact["jobs"] memory_mb = 512 timeout_s = 900 - environment = merge(local.common_env, { AUDIT_ARCHIVE_BUCKET = aws_s3_bucket.audit_archive.bucket }) + environment = merge(local.common_env, { + AUDIT_ARCHIVE_BUCKET = aws_s3_bucket.audit_archive.bucket + AUDIT_EXPORT_BUCKET = aws_s3_bucket.audit_exports.bucket + }) policy_statements = concat(local.data_api, local.secrets, local.sign, local.crypt, local.email, [{ Effect = "Allow" Action = ["s3:PutObject"] - Resource = "${aws_s3_bucket.audit_archive.arn}/*" + Resource = ["${aws_s3_bucket.audit_archive.arn}/*", "${aws_s3_bucket.audit_exports.arn}/*"] }]) } @@ -436,6 +475,7 @@ locals { schedules = { "entitlement-sweep" = "rate(5 minutes)" "webhook-dispatch" = "rate(1 minute)" + "audit-exports" = "rate(1 minute)" "erasure-cascade" = "rate(1 hour)" "webhook-health" = "rate(1 hour)" "subscription-access" = "rate(1 hour)" diff --git a/internal/api/gen/gen.go b/internal/api/gen/gen.go index 4513122..6881bec 100644 --- a/internal/api/gen/gen.go +++ b/internal/api/gen/gen.go @@ -350,6 +350,7 @@ const ( AuditActionApplicationCreated AuditAction = "application.created" AuditActionApplicationReviewStatusChanged AuditAction = "application.review_status_changed" AuditActionApplicationUpdated AuditAction = "application.updated" + AuditActionAuditExported AuditAction = "audit.exported" AuditActionDeviceDeleted AuditAction = "device.deleted" AuditActionEntitlementDeleted AuditAction = "entitlement.deleted" AuditActionEntitlementDisabled AuditAction = "entitlement.disabled" @@ -419,6 +420,8 @@ func (e AuditAction) Valid() bool { return true case AuditActionApplicationUpdated: return true + case AuditActionAuditExported: + return true case AuditActionDeviceDeleted: return true case AuditActionEntitlementDeleted: @@ -520,10 +523,92 @@ func (e AuditAction) Valid() bool { } } +// Defines values for AuditExportFormat. +const ( + AuditExportFormatCsv AuditExportFormat = "csv" + AuditExportFormatNdjson AuditExportFormat = "ndjson" +) + +// Valid indicates whether the value is a known member of the AuditExportFormat enum. +func (e AuditExportFormat) Valid() bool { + switch e { + case AuditExportFormatCsv: + return true + case AuditExportFormatNdjson: + return true + default: + return false + } +} + +// Defines values for AuditExportStatus. +const ( + AuditExportStatusFailed AuditExportStatus = "failed" + AuditExportStatusPending AuditExportStatus = "pending" + AuditExportStatusReady AuditExportStatus = "ready" + AuditExportStatusRunning AuditExportStatus = "running" +) + +// Valid indicates whether the value is a known member of the AuditExportStatus enum. +func (e AuditExportStatus) Valid() bool { + switch e { + case AuditExportStatusFailed: + return true + case AuditExportStatusPending: + return true + case AuditExportStatusReady: + return true + case AuditExportStatusRunning: + return true + default: + return false + } +} + +// Defines values for AuditExportRequestActorType. +const ( + AuditExportRequestActorTypeApiKey AuditExportRequestActorType = "api_key" + AuditExportRequestActorTypeSystem AuditExportRequestActorType = "system" + AuditExportRequestActorTypeUser AuditExportRequestActorType = "user" +) + +// Valid indicates whether the value is a known member of the AuditExportRequestActorType enum. +func (e AuditExportRequestActorType) Valid() bool { + switch e { + case AuditExportRequestActorTypeApiKey: + return true + case AuditExportRequestActorTypeSystem: + return true + case AuditExportRequestActorTypeUser: + return true + default: + return false + } +} + +// Defines values for AuditExportRequestFormat. +const ( + AuditExportRequestFormatCsv AuditExportRequestFormat = "csv" + AuditExportRequestFormatNdjson AuditExportRequestFormat = "ndjson" +) + +// Valid indicates whether the value is a known member of the AuditExportRequestFormat enum. +func (e AuditExportRequestFormat) Valid() bool { + switch e { + case AuditExportRequestFormatCsv: + return true + case AuditExportRequestFormatNdjson: + return true + default: + return false + } +} + // Defines values for AuditTargetType. const ( AuditTargetTypeApiKey AuditTargetType = "api_key" AuditTargetTypeApplication AuditTargetType = "application" + AuditTargetTypeAuditExport AuditTargetType = "audit_export" AuditTargetTypeDevice AuditTargetType = "device" AuditTargetTypeEntitlement AuditTargetType = "entitlement" AuditTargetTypeOrganization AuditTargetType = "organization" @@ -544,6 +629,8 @@ func (e AuditTargetType) Valid() bool { return true case AuditTargetTypeApplication: return true + case AuditTargetTypeAuditExport: + return true case AuditTargetTypeDevice: return true case AuditTargetTypeEntitlement: @@ -2402,6 +2489,62 @@ type AuditEventPage struct { Page PageInfo `json:"page"` } +// AuditExport defines model for AuditExport. +type AuditExport struct { + CompletedAt *time.Time `json:"completed_at,omitempty"` + CreatedAt *time.Time `json:"created_at,omitempty"` + + // DownloadUrl Set once ready. Signed, valid for 15 minutes; fetch the export again for a fresh one. + DownloadUrl *string `json:"download_url,omitempty"` + DownloadUrlExpiresAt *time.Time `json:"download_url_expires_at,omitempty"` + + // Error Set when failed, e.g. too_many_rows. + Error *string `json:"error,omitempty"` + + // ExpiresAt When the file and this record are deleted: 7 days after the request. + ExpiresAt *time.Time `json:"expires_at,omitempty"` + + // Filters The filters as requested, with target_user_id me resolved to the caller's id. + Filters *map[string]interface{} `json:"filters,omitempty"` + Format *AuditExportFormat `json:"format,omitempty"` + Id *string `json:"id,omitempty"` + + // RowCount Set once ready. + RowCount *int `json:"row_count,omitempty"` + Status *AuditExportStatus `json:"status,omitempty"` + TestMode *bool `json:"test_mode,omitempty"` +} + +// AuditExportFormat defines model for AuditExport.Format. +type AuditExportFormat string + +// AuditExportStatus defines model for AuditExport.Status. +type AuditExportStatus string + +// AuditExportRequest The same filters as GET /v1/audit-events, as body fields, plus format. +type AuditExportRequest struct { + Action *[]AuditAction `json:"action,omitempty"` + ActorId *string `json:"actor_id,omitempty"` + ActorType *AuditExportRequestActorType `json:"actor_type,omitempty"` + ApplicationId *string `json:"application_id,omitempty"` + Format *AuditExportRequestFormat `json:"format,omitempty"` + OrganizationId *string `json:"organization_id,omitempty"` + Since *time.Time `json:"since,omitempty"` + TargetId *string `json:"target_id,omitempty"` + TargetType *AuditTargetType `json:"target_type,omitempty"` + + // TargetUserId A User id, or me. + TargetUserId *string `json:"target_user_id,omitempty"` + TenantId *string `json:"tenant_id,omitempty"` + Until *time.Time `json:"until,omitempty"` +} + +// AuditExportRequestActorType defines model for AuditExportRequest.ActorType. +type AuditExportRequestActorType string + +// AuditExportRequestFormat defines model for AuditExportRequest.Format. +type AuditExportRequestFormat string + // AuditTargetType defines model for AuditTargetType. type AuditTargetType string @@ -3567,6 +3710,12 @@ type AuditListParams struct { // AuditListParamsActorType defines parameters for AuditList. type AuditListParamsActorType string +// AuditCreateExportParams defines parameters for AuditCreateExport. +type AuditCreateExportParams struct { + // IdempotencyKey Client-generated, 1–255 characters; a UUID v4 is recommended. Remembered 24 hours per caller. + IdempotencyKey IdempotencyKey `json:"Idempotency-Key"` +} + // AuthCreateAppTokenJSONBody defines parameters for AuthCreateAppToken. type AuthCreateAppTokenJSONBody struct { ApplicationId string `json:"application_id"` @@ -4193,6 +4342,9 @@ type PushCredentialsCreateJSONRequestBody = PushCredentialCreate // PushCredentialsUpdateJSONRequestBody defines body for PushCredentialsUpdate for application/json ContentType. type PushCredentialsUpdateJSONRequestBody = PushCredentialUpdate +// AuditCreateExportJSONRequestBody defines body for AuditCreateExport for application/json ContentType. +type AuditCreateExportJSONRequestBody = AuditExportRequest + // AuthCreateAppTokenJSONRequestBody defines body for AuthCreateAppToken for application/json ContentType. type AuthCreateAppTokenJSONRequestBody AuthCreateAppTokenJSONBody @@ -4712,6 +4864,12 @@ type ServerInterface interface { // AuditGet Fetch one Audit Event // (GET /audit-events/{id}) AuditGet(w http.ResponseWriter, r *http.Request, id string) + // AuditCreateExport Export Audit Events to a file (asynchronous) + // (POST /audit-exports) + AuditCreateExport(w http.ResponseWriter, r *http.Request, params AuditCreateExportParams) + // AuditGetExport Fetch an export's status, and its download link once ready + // (GET /audit-exports/{id}) + AuditGetExport(w http.ResponseWriter, r *http.Request, id string) // AuthCreateAppToken Mint a per-app JWT from a platform session (embedded login) // (POST /auth/app-tokens) AuthCreateAppToken(w http.ResponseWriter, r *http.Request) @@ -5916,6 +6074,77 @@ func (siw *ServerInterfaceWrapper) AuditGet(w http.ResponseWriter, r *http.Reque handler.ServeHTTP(w, r) } +// AuditCreateExport operation middleware +func (siw *ServerInterfaceWrapper) AuditCreateExport(w http.ResponseWriter, r *http.Request) { + + var err error + _ = err + + // Parameter object where we will unmarshal all parameters from the context + var params AuditCreateExportParams + + headers := r.Header + + // ------------- Required header parameter "Idempotency-Key" ------------- + if valueList, found := headers[http.CanonicalHeaderKey("Idempotency-Key")]; found { + var IdempotencyKey IdempotencyKey + n := len(valueList) + if n != 1 { + siw.ErrorHandlerFunc(w, r, &TooManyValuesForParamError{ParamName: "Idempotency-Key", Count: n}) + return + } + + err = runtime.BindStyledParameterWithOptions("simple", "Idempotency-Key", valueList[0], &IdempotencyKey, runtime.BindStyledParameterOptions{ParamLocation: runtime.ParamLocationHeader, Explode: false, Required: true, Type: "string", Format: ""}) + if err != nil { + siw.ErrorHandlerFunc(w, r, &InvalidParamFormatError{ParamName: "Idempotency-Key", Err: err}) + return + } + + params.IdempotencyKey = IdempotencyKey + + } else { + err := fmt.Errorf("Header parameter Idempotency-Key is required, but not found") + siw.ErrorHandlerFunc(w, r, &RequiredHeaderError{ParamName: "Idempotency-Key", Err: err}) + return + } + + handler := http.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + siw.Handler.AuditCreateExport(w, r, params) + })) + + for _, middleware := range siw.HandlerMiddlewares { + handler = middleware(handler) + } + + handler.ServeHTTP(w, r) +} + +// AuditGetExport operation middleware +func (siw *ServerInterfaceWrapper) AuditGetExport(w http.ResponseWriter, r *http.Request) { + + var err error + _ = err + + // ------------- Path parameter "id" ------------- + var id string + + err = runtime.BindStyledParameterWithOptions("simple", "id", r.PathValue("id"), &id, runtime.BindStyledParameterOptions{ParamLocation: runtime.ParamLocationPath, Explode: false, Required: true, Type: "string", Format: "", ValueIsUnescaped: true}) + if err != nil { + siw.ErrorHandlerFunc(w, r, &InvalidParamFormatError{ParamName: "id", Err: err}) + return + } + + handler := http.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + siw.Handler.AuditGetExport(w, r, id) + })) + + for _, middleware := range siw.HandlerMiddlewares { + handler = middleware(handler) + } + + handler.ServeHTTP(w, r) +} + // AuthCreateAppToken operation middleware func (siw *ServerInterfaceWrapper) AuthCreateAppToken(w http.ResponseWriter, r *http.Request) { @@ -10266,6 +10495,8 @@ func HandlerWithOptions(si ServerInterface, options StdHTTPServerOptions) http.H m.HandleFunc(http.MethodPost+" "+options.BaseURL+"/organizations/{id}/billing/portal-sessions", wrapper.BillingCreateOrganizationPortalSession) m.HandleFunc(http.MethodGet+" "+options.BaseURL+"/audit-events", wrapper.AuditList) m.HandleFunc(http.MethodGet+" "+options.BaseURL+"/audit-events/{id}", wrapper.AuditGet) + m.HandleFunc(http.MethodPost+" "+options.BaseURL+"/audit-exports", wrapper.AuditCreateExport) + m.HandleFunc(http.MethodGet+" "+options.BaseURL+"/audit-exports/{id}", wrapper.AuditGetExport) m.HandleFunc(http.MethodGet+" "+options.BaseURL+"/webhooks", wrapper.WebhooksList) m.HandleFunc(http.MethodPost+" "+options.BaseURL+"/webhooks", wrapper.WebhooksCreate) m.HandleFunc(http.MethodDelete+" "+options.BaseURL+"/webhooks/{id}", wrapper.WebhooksDelete) diff --git a/internal/core/audit.go b/internal/core/audit.go index 8b64558..7d5bac8 100644 --- a/internal/core/audit.go +++ b/internal/core/audit.go @@ -44,81 +44,114 @@ func (c *Call) ownsTenant(t tenantRow) (bool, error) { return c.isOrgAdmin(*t.OwnerOrganizationID) } +// auditForced is the scope a caller without audit.view is confined to: at +// most one field is set, and none for audit.view. +type auditForced struct { + ApplicationID string `json:"application_id,omitempty"` + TargetUserID string `json:"target_user_id,omitempty"` + TenantID string `json:"tenant_id,omitempty"` +} + // auditScope returns the forced scope for the caller, or 403. -func (c *Call) auditScope(targetUser, tenant *string) (where []string, args db.Args, err error) { - args = db.Args{} +func (c *Call) auditScope(targetUser, tenant *string) (auditForced, error) { if c.can("audit.view") { - return nil, args, nil + return auditForced{}, nil } if app, ok := c.p.AppKey(); ok { if !c.p.Perms["audit.view"] { - return nil, nil, httpx.Forbidden("audit.view") + return auditForced{}, httpx.Forbidden("audit.view") } - args["forced_app"] = app - return []string{"a.application_id = :forced_app"}, args, nil + return auditForced{ApplicationID: app}, nil } if c.p.IsUser() && targetUser != nil && *targetUser == c.p.UserID { - args["forced_user"] = c.p.UserID - return []string{"a.target_user_id = :forced_user"}, args, nil + return auditForced{TargetUserID: c.p.UserID}, nil } if tenant != nil { t, err := c.loadTenant(*tenant) if err == nil { if owner, err := c.ownsTenant(t); err != nil { - return nil, nil, err + return auditForced{}, err } else if owner { - args["forced_tenant"] = t.ID - return []string{"a.tenant_id = :forced_tenant"}, args, nil + return auditForced{TenantID: t.ID}, nil } } } - return nil, nil, httpx.Forbidden("audit.view") + return auditForced{}, httpx.Forbidden("audit.view") +} + +// auditFilters are GET /v1/audit-events' filters, and an export's. +type auditFilters struct { + TargetUserID *string `json:"target_user_id,omitempty"` + TargetType *string `json:"target_type,omitempty"` + TargetID *string `json:"target_id,omitempty"` + ActorID *string `json:"actor_id,omitempty"` + ActorType *string `json:"actor_type,omitempty"` + Action []string `json:"action,omitempty"` + ApplicationID *string `json:"application_id,omitempty"` + OrganizationID *string `json:"organization_id,omitempty"` + TenantID *string `json:"tenant_id,omitempty"` + Since *time.Time `json:"since,omitempty"` + Until *time.Time `json:"until,omitempty"` +} + +// auditWhere is the where clause for a forced scope and filters. +func auditWhere(forced auditForced, f auditFilters) ([]string, db.Args) { + var where []string + args := db.Args{} + add := func(cond, key string, v *string) { + if v != nil && *v != "" { + where = append(where, cond) + args[key] = *v + } + } + add("a.application_id = :forced_app", "forced_app", &forced.ApplicationID) + add("a.target_user_id = :forced_user", "forced_user", &forced.TargetUserID) + add("a.tenant_id = :forced_tenant", "forced_tenant", &forced.TenantID) + add("a.target_user_id = :tu", "tu", f.TargetUserID) + add("a.target_type = :tt", "tt", f.TargetType) + add("a.target_id = :ti", "ti", f.TargetID) + add("a.actor_id = :ai", "ai", f.ActorID) + add("a.actor_type = :at", "at", f.ActorType) + add("a.application_id = :app", "app", f.ApplicationID) + add("a.organization_id = :org", "org", f.OrganizationID) + add("a.tenant_id = :tnt", "tnt", f.TenantID) + if len(f.Action) > 0 { + where = append(where, "a.action = any(:acts::text[])") + args["acts"] = db.TextArray(f.Action) + } + if f.Since != nil { + where = append(where, "a.timestamp >= :since::timestamptz") + args["since"] = db.Time(*f.Since) + } + if f.Until != nil { + where = append(where, "a.timestamp < :until::timestamptz") + args["until"] = db.Time(*f.Until) + } + return where, args } func (s *Server) AuditList(w http.ResponseWriter, r *http.Request, params gen.AuditListParams) { s.serve(w, r, opts{op: "audit.list"}, func(c *Call) error { - var targetUser *string + f := auditFilters{TargetType: enumStr(params.TargetType), TargetID: params.TargetId, ActorID: params.ActorId, + ActorType: enumStr(params.ActorType), ApplicationID: params.ApplicationId, OrganizationID: params.OrganizationId, + TenantID: params.TenantId, Since: params.Since, Until: params.Until} if params.TargetUserId != nil { uid, err := c.selfID(*params.TargetUserId) if err != nil { return err } - targetUser = &uid + f.TargetUserID = &uid } - where, args, err := c.auditScope(targetUser, params.TenantId) - if err != nil { - return err - } - add := func(cond, key string, v *string) { - if v != nil { - where = append(where, cond) - args[key] = *v + if params.Action != nil { + for _, a := range *params.Action { + f.Action = append(f.Action, string(a)) } } - add("a.target_user_id = :tu", "tu", targetUser) - add("a.target_type = :tt", "tt", enumStr(params.TargetType)) - add("a.target_id = :ti", "ti", params.TargetId) - add("a.actor_id = :ai", "ai", params.ActorId) - add("a.actor_type = :at", "at", enumStr(params.ActorType)) - add("a.application_id = :app", "app", params.ApplicationId) - add("a.organization_id = :org", "org", params.OrganizationId) - add("a.tenant_id = :tnt", "tnt", params.TenantId) - if params.Action != nil && len(*params.Action) > 0 { - acts := make([]string, len(*params.Action)) - for i, a := range *params.Action { - acts[i] = string(a) - } - where = append(where, "a.action = any(:acts::text[])") - args["acts"] = db.TextArray(acts) - } - if params.Since != nil { - where = append(where, "a.timestamp >= :since::timestamptz") - args["since"] = db.Time(*params.Since) - } - if params.Until != nil { - where = append(where, "a.timestamp < :until::timestamptz") - args["until"] = db.Time(*params.Until) + forced, err := c.auditScope(f.TargetUserID, params.TenantId) + if err != nil { + return err } + where, args := auditWhere(forced, f) lim := httpx.Limit(params.Limit) args["lim"] = int64(lim + 1) filters := httpx.Filters(c.r) @@ -155,11 +188,11 @@ func (s *Server) AuditGet(w http.ResponseWriter, r *http.Request, id string) { } else if err != nil { return err } - where, args, err := c.auditScope(ev.TargetUserID, ev.TenantID) + forced, err := c.auditScope(ev.TargetUserID, ev.TenantID) if err != nil { return httpx.NotFoundResource("audit_event") } - if len(where) > 0 { + if where, args := auditWhere(forced, auditFilters{}); len(where) > 0 { args["id"] = id rows, err := c.q.Query(c.ctx, `select to_jsonb(a.id) from audit_events a where a.id = :id and `+ strings.Join(where, " and "), args) diff --git a/internal/core/audit_exports.go b/internal/core/audit_exports.go new file mode 100644 index 0000000..f95234f --- /dev/null +++ b/internal/core/audit_exports.go @@ -0,0 +1,347 @@ +package core + +import ( + "context" + "encoding/csv" + "encoding/json" + "io" + "net/http" + "os" + "strings" + "time" + + "github.com/CompositeCode/substratalapps.com/internal/api/gen" + "github.com/CompositeCode/substratalapps.com/internal/db" + "github.com/CompositeCode/substratalapps.com/internal/httpx" + "github.com/CompositeCode/substratalapps.com/internal/ids" +) + +// ExportStore holds finished Audit Event exports: a private S3 bucket in +// AWS (objects deleted after 7 days by a lifecycle rule), a local +// directory otherwise. +type ExportStore interface { + Put(ctx context.Context, key, contentType string, body io.ReadSeeker) error + // URL returns a link that downloads key for ttl. + URL(ctx context.Context, key string, ttl time.Duration) (string, error) +} + +const ( + auditExportTTL = 7 * 24 * time.Hour + auditExportURLTTL = 15 * time.Minute + auditExportMaxRows = 1_000_000 + auditExportPageSize = 1000 + // A running export whose worker died is picked up again after this. + auditExportStale = 20 * time.Minute +) + +type auditExportRow struct { + ID string `json:"id"` + RequestedByType string `json:"requested_by_type"` + RequestedByID string `json:"requested_by_id"` + Scope auditForced `json:"scope"` + Filters json.RawMessage `json:"filters"` + Format string `json:"format"` + Status string `json:"status"` + RowCount *int `json:"row_count"` + ObjectKey *string `json:"object_key"` + Error *string `json:"error"` + TestMode bool `json:"test_mode"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at"` + ExpiresAt time.Time `json:"expires_at"` +} + +// auditExport is the API AuditExport object. +type auditExport struct { + ID string `json:"id"` + Status string `json:"status"` + Format string `json:"format"` + Filters json.RawMessage `json:"filters"` + RowCount *int `json:"row_count"` + DownloadURL *string `json:"download_url"` + DownloadURLExpiresAt *time.Time `json:"download_url_expires_at"` + Error *string `json:"error"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at"` + ExpiresAt time.Time `json:"expires_at"` + TestMode bool `json:"test_mode"` +} + +// exportObject builds the API object, signing a fresh download link when +// the file is ready. +func (c *Call) exportObject(x auditExportRow) (auditExport, error) { + out := auditExport{ID: x.ID, Status: x.Status, Format: x.Format, Filters: x.Filters, RowCount: x.RowCount, + Error: x.Error, CreatedAt: x.CreatedAt, CompletedAt: x.CompletedAt, ExpiresAt: x.ExpiresAt, TestMode: x.TestMode} + if x.Status == "ready" && x.ObjectKey != nil { + u, err := c.s.Exports.URL(c.ctx, *x.ObjectKey, auditExportURLTTL) + if err != nil { + return out, err + } + exp := c.now.Add(auditExportURLTTL) + out.DownloadURL, out.DownloadURLExpiresAt = &u, &exp + } + return out, nil +} + +func exportsUnavailable() error { + return httpx.E(503, "service_unavailable", "Audit exports aren't configured in this environment.") +} + +// ── POST /v1/audit-exports ───────────────────────────────────────────── + +func (s *Server) AuditCreateExport(w http.ResponseWriter, r *http.Request, params gen.AuditCreateExportParams) { + s.serve(w, r, opts{op: "audit.createExport", idem: idemRequired}, func(c *Call) error { + if c.s.Exports == nil { + return exportsUnavailable() + } + var in struct { + Format *string `json:"format"` + auditFilters + } + if err := c.Decode(&in); err != nil { + return err + } + var f httpx.Fields + format := "ndjson" + if in.Format != nil { + format = *in.Format + if format != "ndjson" && format != "csv" { + f.Enum("format", "ndjson", "csv") + } + } + if in.ActorType != nil { + switch *in.ActorType { + case "user", "api_key", "system": + default: + f.Enum("actor_type", "user", "api_key", "system") + } + } + if len(in.Action) > 100 { + f.Max("action", "too_long", 100) + } + if in.Since != nil && in.Until != nil && !in.Since.Before(*in.Until) { + f.Add("until", "out_of_range").Message = "until must be after since" + } + if err := f.Err(); err != nil { + return err + } + filters := in.auditFilters + if filters.TargetUserID != nil { + uid, err := c.selfID(*filters.TargetUserID) + if err != nil { + return err + } + filters.TargetUserID = &uid + } + forced, err := c.auditScope(filters.TargetUserID, filters.TenantID) + if err != nil { + return err + } + typ, by := c.actor() + if err := c.hit("auditexport:"+by, perDay(20), false); err != nil { + return err + } + var x auditExportRow + if err := db.One(c.ctx, c.q, &x, `insert into audit_exports (id, requested_by_type, requested_by_id, scope, filters, + format, test_mode, expires_at) + values (:id, :bt, :bi, :scope::jsonb, :filters::jsonb, :fmt, :tm, :exp::timestamptz) + returning to_jsonb(audit_exports)`, + db.Args{"id": ids.New(ids.AuditExport), "bt": typ, "bi": by, "scope": db.JSON(forced), "filters": db.JSON(filters), + "fmt": format, "tm": c.testMode(), "exp": db.Time(c.now.Add(auditExportTTL))}); err != nil { + return err + } + out, err := c.exportObject(x) + if err != nil { + return err + } + return c.Accepted(out) + }) +} + +// ── GET /v1/audit-exports/{id} ───────────────────────────────────────── + +func (s *Server) AuditGetExport(w http.ResponseWriter, r *http.Request, id string) { + s.serve(w, r, opts{op: "audit.getExport"}, func(c *Call) error { + if c.s.Exports == nil { + return exportsUnavailable() + } + var x auditExportRow + err := db.One(c.ctx, c.q, &x, `select to_jsonb(x) from audit_exports x where id = :id`, db.Args{"id": id}) + typ, by := c.actor() + if err == db.ErrNotFound || (err == nil && !(x.RequestedByType == typ && x.RequestedByID == by) && !c.can("audit.view")) { + return httpx.NotFoundResource("audit_export") + } + if err != nil { + return err + } + out, err := c.exportObject(x) + if err != nil { + return err + } + return c.OK(out) + }) +} + +// ── Worker (jobs: audit-exports, every minute) ───────────────────────── + +// RunAuditExports writes every due export, one at a time, in both modes. +func (s *Server) RunAuditExports(ctx context.Context) (done int, err error) { + if s.Exports == nil { + return 0, nil + } + err = eachMode(func(tm bool) error { + for { + var claimed []auditExportRow + if err := s.system(ctx, tm, db.Settings{}, func(c *Call) error { + var err error + claimed, err = db.All[auditExportRow](ctx, c.q, `update audit_exports set status = 'running', started_at = now() + where id = (select id from audit_exports + where status = 'pending' or (status = 'running' and started_at < now() - make_interval(secs => :stale)) + order by created_at limit 1 for update skip locked) + returning to_jsonb(audit_exports)`, db.Args{"stale": auditExportStale.Seconds()}) + return err + }); err != nil { + return err + } + if len(claimed) == 0 { + return nil + } + if err := s.runAuditExport(ctx, tm, claimed[0]); err != nil { + return err + } + done++ + } + }) + return +} + +// runAuditExport writes one export's file and marks it ready, or failed. +func (s *Server) runAuditExport(ctx context.Context, tm bool, x auditExportRow) error { + var filters auditFilters + if err := json.Unmarshal(x.Filters, &filters); err != nil { + return s.failAuditExport(ctx, tm, x, "internal_error") + } + file, err := os.CreateTemp("", "audit-export-*") + if err != nil { + return err + } + defer func() { _ = file.Close(); _ = os.Remove(file.Name()) }() + + var write func(auditEvent) error + var flush func() error + contentType, ext := "application/x-ndjson", "ndjson" + if x.Format == "csv" { + contentType, ext = "text/csv", "csv" + cw := csv.NewWriter(file) + if err := cw.Write(auditCSVHeader); err != nil { + return err + } + write = func(e auditEvent) error { return cw.Write(auditCSVRow(e)) } + flush = func() error { cw.Flush(); return cw.Error() } + } else { + enc := json.NewEncoder(file) + write = func(e auditEvent) error { return enc.Encode(e) } + flush = func() error { return nil } + } + + // Page through in (timestamp, id) order, one short transaction per + // page, so a large export never holds everything in memory. + count := 0 + var after *auditEvent + for { + var page []auditEvent + if err := s.system(ctx, tm, db.Settings{}, func(c *Call) error { + where, args := auditWhere(x.Scope, filters) + if after != nil { + where = append(where, "(a.timestamp, a.id) > (:ps::timestamptz, :pi)") + args["ps"], args["pi"] = db.Time(after.Timestamp), after.ID + } + if len(where) == 0 { + where = []string{"true"} + } + args["lim"] = int64(auditExportPageSize) + var err error + page, err = db.All[auditEvent](ctx, c.q, auditSQL+" where "+strings.Join(where, " and ")+ + " order by a.timestamp, a.id limit :lim", args) + return err + }); err != nil { + return err + } + for _, e := range page { + if err := write(e); err != nil { + return err + } + } + count += len(page) + if count > auditExportMaxRows { + return s.failAuditExport(ctx, tm, x, "too_many_rows") + } + if len(page) < auditExportPageSize { + break + } + after = &page[len(page)-1] + } + if err := flush(); err != nil { + return err + } + if _, err := file.Seek(0, io.SeekStart); err != nil { + return err + } + prefix := "live" + if tm { + prefix = "test" + } + key := "exports/" + prefix + "/" + x.ID + "." + ext + if err := s.Exports.Put(ctx, key, contentType, file); err != nil { + return err + } + return s.system(ctx, tm, db.Settings{}, func(c *Call) error { + if _, err := c.q.Exec(ctx, `update audit_exports set status = 'ready', row_count = :n, object_key = :k, + completed_at = now() where id = :id`, db.Args{"id": x.ID, "n": int64(count), "k": key}); err != nil { + return err + } + return c.audit(auditExportedEvent(x, count)) + }) +} + +func (s *Server) failAuditExport(ctx context.Context, tm bool, x auditExportRow, reason string) error { + return s.system(ctx, tm, db.Settings{}, func(c *Call) error { + _, err := c.q.Exec(ctx, `update audit_exports set status = 'failed', error = :e, completed_at = now() where id = :id`, + db.Args{"id": x.ID, "e": reason}) + return err + }) +} + +// auditExportedEvent records who exported what, in the export's own scope, +// so it shows up wherever the exported events would. +func auditExportedEvent(x auditExportRow, count int) Audit { + a := Audit{Action: "audit.exported", TargetType: "audit_export", TargetID: x.ID, + After: map[string]any{"filters": x.Filters, "format": x.Format, "row_count": count}, + ActorType: x.RequestedByType, ActorID: x.RequestedByID} + if v := x.Scope.ApplicationID; v != "" { + a.ApplicationID = &v + } + if v := x.Scope.TargetUserID; v != "" { + a.TargetUserID = &v + } + if v := x.Scope.TenantID; v != "" { + a.TenantID = &v + } + return a +} + +var auditCSVHeader = []string{"id", "timestamp", "action", "actor_type", "actor_id", "target_type", "target_id", + "target_user_id", "application_id", "organization_id", "tenant_id", "request_id", "before", "after"} + +func auditCSVRow(e auditEvent) []string { + return []string{e.ID, e.Timestamp.UTC().Format(time.RFC3339Nano), e.Action, e.Actor.Type, e.Actor.ID, + e.Target.Type, e.Target.ID, deref(e.TargetUserID), deref(e.ApplicationID), deref(e.OrganizationID), + deref(e.TenantID), deref(e.RequestID), jsonCell(e.Before), jsonCell(e.After)} +} + +func jsonCell(raw json.RawMessage) string { + if len(raw) == 0 || string(raw) == "null" { + return "" + } + return string(raw) +} diff --git a/internal/core/audit_exports_test.go b/internal/core/audit_exports_test.go new file mode 100644 index 0000000..49cafd3 --- /dev/null +++ b/internal/core/audit_exports_test.go @@ -0,0 +1,119 @@ +package core_test + +import ( + "bufio" + "bytes" + "context" + "encoding/csv" + "encoding/json" + "testing" + + "github.com/CompositeCode/substratalapps.com/internal/testenv" +) + +// runExports runs the export worker and fetches the export, which must be +// ready. +func runExport(t *testing.T, e *testenv.Env, token, id string) map[string]any { + t.Helper() + if _, err := e.Core.RunAuditExports(context.Background()); err != nil { + t.Fatalf("worker: %v", err) + } + x := e.Get("/v1/audit-exports/"+id, token).Expect(200).JSON() + if x["status"] != "ready" || x["download_url"] == nil { + t.Fatalf("export = %v", x) + } + return x +} + +func download(t *testing.T, e *testenv.Env, x map[string]any) []byte { + t.Helper() + b, ok := e.Exports.Download(x["download_url"].(string)) + if !ok { + t.Fatalf("no file behind %v", x["download_url"]) + } + return b +} + +// API Reference → Audit → Exporting. +func TestAuditExport(t *testing.T) { + e := testenv.New(t) + u := e.Signup() + app := e.App(system, nil) + e.Grant(e.Admin, u.ID, app) + e.Patch("/v1/users/"+u.ID+"/profile", e.Admin, map[string]any{"display_name": "Renamed"}).Expect(200) + want := e.Get("/v1/audit-events?target_user_id="+u.ID, e.Admin).Expect(200).JSON()["data"].([]any) + if len(want) < 3 { + t.Fatalf("only %d events to export", len(want)) + } + + // NDJSON, by audit.view: every matching event, oldest first. + req := e.PostIdem("/v1/audit-exports", e.Admin, testenv.Unique(), map[string]any{"target_user_id": u.ID}).Expect(202).JSON() + if req["status"] != "pending" || req["download_url"] != nil || req["format"] != "ndjson" { + t.Fatalf("request = %v", req) + } + id := req["id"].(string) + x := runExport(t, e, e.Admin, id) + var got []map[string]any + sc := bufio.NewScanner(bytes.NewReader(download(t, e, x))) + for sc.Scan() { + var ev map[string]any + if err := json.Unmarshal(sc.Bytes(), &ev); err != nil { + t.Fatalf("line %q: %v", sc.Text(), err) + } + got = append(got, ev) + } + if len(got) != len(want) || int(x["row_count"].(float64)) != len(want) { + t.Fatalf("exported %d rows (row_count %v), want %d", len(got), x["row_count"], len(want)) + } + if got[0]["id"] != want[len(want)-1].(map[string]any)["id"] { + t.Fatal("export should be oldest first") + } + + // The export is itself audited, as its requester. + exported := e.Get("/v1/audit-events?action=audit.exported&target_id="+id, e.Admin).Expect(200).JSON()["data"].([]any) + if len(exported) != 1 || exported[0].(map[string]any)["actor"].(map[string]any)["id"] != e.AdminID { + t.Fatalf("audit.exported = %v", exported) + } + + // Idempotent: the same key returns the same export. + key := testenv.Unique() + a := e.PostIdem("/v1/audit-exports", e.Admin, key, map[string]any{"format": "csv", "target_user_id": u.ID}).Expect(202).Str("id") + b := e.PostIdem("/v1/audit-exports", e.Admin, key, map[string]any{"format": "csv", "target_user_id": u.ID}).Expect(202).Str("id") + if a != b { + t.Fatalf("idempotent request made two exports: %s, %s", a, b) + } + // CSV: a header, then one row per event. + rows, err := csv.NewReader(bytes.NewReader(download(t, e, runExport(t, e, e.Admin, a)))).ReadAll() + if err != nil || len(rows) != len(want)+1 || rows[0][0] != "id" || rows[0][2] != "action" { + t.Fatalf("csv: %d rows, %v", len(rows), err) + } + + // Validation. + e.PostIdem("/v1/audit-exports", e.Admin, testenv.Unique(), map[string]any{"format": "xml"}).ExpectErr(422, "validation_failed") + + // A User exports only their own trail, and sees only their own exports. + e.PostIdem("/v1/audit-exports", u.Token, testenv.Unique(), map[string]any{}).ExpectErr(403, "forbidden") + own := e.PostIdem("/v1/audit-exports", u.Token, testenv.Unique(), map[string]any{"target_user_id": "me"}).Expect(202).Str("id") + if n := int(runExport(t, e, u.Token, own)["row_count"].(float64)); n < len(want) { + t.Fatalf("own trail exported %d rows, want at least %d", n, len(want)) + } + e.Get("/v1/audit-exports/"+id, u.Token).ExpectErr(404, "audit_export_not_found") + + // An app-confined key is forced to its own Application. + appKey := e.Post("/v1/api-keys", e.Admin, map[string]any{"name": "auditor", "scope": app, + "permissions": []string{"audit.view"}}).Expect(201).Str("secret") + scoped := e.PostIdem("/v1/audit-exports", appKey, testenv.Unique(), map[string]any{"target_user_id": u.ID}).Expect(202).Str("id") + sc = bufio.NewScanner(bytes.NewReader(download(t, e, runExport(t, e, appKey, scoped)))) + n := 0 + for sc.Scan() { + var ev map[string]any + _ = json.Unmarshal(sc.Bytes(), &ev) + if ev["application_id"] != app { + t.Fatalf("app key exported another Application's event: %v", ev) + } + n++ + } + if n == 0 { + t.Fatal("app key exported nothing") + } +} diff --git a/internal/core/server.go b/internal/core/server.go index 0014500..36ee953 100644 --- a/internal/core/server.go +++ b/internal/core/server.go @@ -66,6 +66,7 @@ type Deps struct { Notifier Notifier Billing Billing // nil disables checkout/portal (503 billing_unavailable) Push push.Sender // nil logs pushes instead of sending them (PUSH_BACKEND=log) + Exports ExportStore // nil disables audit exports (503) Cursors httpx.Cursors Log *slog.Logger diff --git a/internal/core/system.go b/internal/core/system.go index b700700..b07e1bb 100644 --- a/internal/core/system.go +++ b/internal/core/system.go @@ -345,6 +345,7 @@ func (s *Server) Cleanup(ctx context.Context) error { `delete from webhook_events e where created_at < now() - interval '30 days' and not exists (select 1 from webhook_deliveries d where d.event_id = e.id)`, `delete from stripe_events where received_at < now() - interval '90 days'`, + `delete from audit_exports where expires_at < now()`, } { if _, err := c.q.Exec(ctx, q, nil); err != nil { return errf("cleanup %q: %w", strings.Fields(q)[2], err) diff --git a/internal/ids/ids.go b/internal/ids/ids.go index 0720d8f..dfaae29 100644 --- a/internal/ids/ids.go +++ b/internal/ids/ids.go @@ -34,6 +34,7 @@ const ( PushCredential = "pcr_" Notification = "ntf_" NotificationDelivery = "ndl_" + AuditExport = "aex_" ) // Credential prefixes. Never logged, never echoed after issuance. diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index f1ac082..854e4b3 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -56,6 +56,12 @@ func (r *Runner) Jobs() map[string]func(context.Context) error { } return r.Base.Deliverer().Run(ctx) }, + // every minute: write requested Audit Event exports + "audit-exports": func(ctx context.Context) error { + n, err := r.Core.RunAuditExports(ctx) + r.Base.Log.Info("audit exports", "written", n) + return err + }, // daily: stale devices, inactive-device deletion, inbox retention "push-maintenance": func(ctx context.Context) error { return r.Core.PushMaintenance(ctx) }, // hourly diff --git a/internal/mcp/tools.json b/internal/mcp/tools.json index 1398a8f..e8ee671 100644 --- a/internal/mcp/tools.json +++ b/internal/mcp/tools.json @@ -974,6 +974,169 @@ "x_needs_idempotency_key": false, "x_accepts_if_match": true }, + { + "name": "substratal_audit_createExport", + "title": "Export Audit Events to a file (asynchronous)", + "description": "Export Audit Events to a file (asynchronous)\n\nExports every Audit Event matching the same filters as GET /v1/audit-events, in the caller's scope there, as NDJSON or CSV. Hot window only. Returns 202 at once; a job writes the file, usually within a minute. Poll GET /v1/audit-exports/{id} until status is ready, then download from download_url. Exports are kept 7 days. Writes audit.exported when the file is ready.\n\nREST: POST /v1/audit-exports", + "inputSchema": { + "additionalProperties": false, + "properties": { + "body": { + "description": "The same filters as GET /v1/audit-events, as body fields, plus format.", + "properties": { + "action": { + "items": { + "enum": [ + "user.created", + "user.updated", + "user.email_changed", + "user.suspended", + "user.reactivated", + "user.deleted", + "user.erasure_requested", + "user.erasure_cancelled", + "user.erased", + "user.password_changed", + "user.password_reset", + "user.mfa_enabled", + "user.mfa_disabled", + "user.mfa_reset", + "audit.exported", + "entitlement.granted", + "entitlement.disabled", + "entitlement.revoked", + "entitlement.expired", + "entitlement.updated", + "entitlement.member_scope_changed", + "entitlement.deleted", + "role.created", + "role.updated", + "role.deleted", + "role.assigned", + "role.removed", + "profile.updated", + "settings.updated", + "application.created", + "application.updated", + "application.review_status_changed", + "organization.created", + "organization.updated", + "organization.member_invited", + "organization.member_added", + "organization.member_removed", + "organization.member_role_changed", + "subscription.created", + "subscription.status_changed", + "subscription.members_changed", + "tenant.tier_change_requested", + "tenant.tier_change_request_updated", + "tenant.tier_changed", + "api_key.created", + "api_key.updated", + "api_key.rotated", + "api_key.revoked", + "api_key.restrict_destructive_disabled", + "webhook.created", + "webhook.updated", + "webhook.deleted", + "webhook.secret_rotated", + "push_credential.created", + "push_credential.updated", + "push_credential.deleted", + "device.deleted" + ], + "type": "string" + }, + "maxItems": 100, + "type": "array" + }, + "actor_id": { + "type": "string" + }, + "actor_type": { + "enum": [ + "user", + "api_key", + "system" + ], + "type": "string" + }, + "application_id": { + "type": "string" + }, + "format": { + "default": "ndjson", + "enum": [ + "ndjson", + "csv" + ], + "type": "string" + }, + "organization_id": { + "type": "string" + }, + "since": { + "format": "date-time", + "type": "string" + }, + "target_id": { + "type": "string" + }, + "target_type": { + "enum": [ + "user", + "entitlement", + "role", + "role_assignment", + "application", + "organization", + "subscription", + "tenant", + "tier_change_request", + "api_key", + "webhook", + "push_credential", + "device", + "audit_export" + ], + "type": "string" + }, + "target_user_id": { + "description": "A User id, or me.", + "type": "string" + }, + "tenant_id": { + "type": "string" + }, + "until": { + "format": "date-time", + "type": "string" + } + }, + "type": "object" + } + }, + "required": [ + "body" + ], + "type": "object" + }, + "annotations": { + "destructiveHint": false, + "idempotentHint": true, + "openWorldHint": false, + "readOnlyHint": false, + "title": "Export Audit Events to a file (asynchronous)" + }, + "x_operation_id": "audit.createExport", + "x_method": "POST", + "x_path": "/v1/audit-exports", + "x_path_params": null, + "x_query_params": null, + "x_has_body": true, + "x_needs_idempotency_key": true, + "x_accepts_if_match": false + }, { "name": "substratal_audit_get", "title": "Fetch one Audit Event", @@ -1008,6 +1171,40 @@ "x_needs_idempotency_key": false, "x_accepts_if_match": false }, + { + "name": "substratal_audit_getExport", + "title": "Fetch an export's status, and its download link once ready", + "description": "Fetch an export's status, and its download link once ready\n\nVisible to whoever requested it, and to audit.view. download_url is freshly signed on every call and valid for 15 minutes.\n\nREST: GET /v1/audit-exports/{id}", + "inputSchema": { + "additionalProperties": false, + "properties": { + "id": { + "type": "string" + } + }, + "required": [ + "id" + ], + "type": "object" + }, + "annotations": { + "destructiveHint": false, + "idempotentHint": true, + "openWorldHint": false, + "readOnlyHint": true, + "title": "Fetch an export's status, and its download link once ready" + }, + "x_operation_id": "audit.getExport", + "x_method": "GET", + "x_path": "/v1/audit-exports/{id}", + "x_path_params": [ + "id" + ], + "x_query_params": null, + "x_has_body": false, + "x_needs_idempotency_key": false, + "x_accepts_if_match": false + }, { "name": "substratal_audit_list", "title": "Query the audit log (hot window only)", @@ -1032,6 +1229,7 @@ "user.mfa_enabled", "user.mfa_disabled", "user.mfa_reset", + "audit.exported", "entitlement.granted", "entitlement.disabled", "entitlement.revoked", @@ -1126,7 +1324,8 @@ "api_key", "webhook", "push_credential", - "device" + "device", + "audit_export" ], "type": "string" }, diff --git a/internal/platform/config/config.go b/internal/platform/config/config.go index e84770f..48039ff 100644 --- a/internal/platform/config/config.go +++ b/internal/platform/config/config.go @@ -50,6 +50,10 @@ type Config struct { // (the local default, so no Apple or Google credentials are needed). PushBackend string + // Audit exports: the private S3 bucket finished exports go to. Empty = + // a local directory (.dev/audit-exports). + AuditExportBucket string + // Stripe. StripeSecretKey string StripeWebhookSecret string @@ -97,6 +101,7 @@ func Load() (Config, error) { EmailFrom: get("EMAIL_FROM", "Substratal "), WebhookQueueURL: os.Getenv("WEBHOOK_QUEUE_URL"), PushBackend: get("PUSH_BACKEND", "log"), + AuditExportBucket: os.Getenv("AUDIT_EXPORT_BUCKET"), StripeSecretKey: os.Getenv("STRIPE_SECRET_KEY"), StripeWebhookSecret: os.Getenv("STRIPE_WEBHOOK_SECRET"), StripePrices: StripePrices{ diff --git a/internal/platform/wire/deps.go b/internal/platform/wire/deps.go index 60e286b..d2bd623 100644 --- a/internal/platform/wire/deps.go +++ b/internal/platform/wire/deps.go @@ -138,6 +138,7 @@ type Base struct { Encrypter auth.Encrypter Email email.Sender Push push.Sender + Exports core.ExportStore } // NewBase builds the shared dependencies. @@ -167,7 +168,12 @@ func NewBase(ctx context.Context) (*Base, error) { if err != nil { return nil, err } - return &Base{Config: cfg, Log: log, DB: d, Signer: signer, Encrypter: enc, Email: mail, Push: sender}, nil + exports, err := Exports(ctx, cfg) + if err != nil { + return nil, err + } + return &Base{Config: cfg, Log: log, DB: d, Signer: signer, Encrypter: enc, Email: mail, Push: sender, + Exports: exports}, nil } // Deliverer is the webhook delivery engine. @@ -197,7 +203,7 @@ func (b *Base) Server(n core.Notifier) *core.Server { } core.AllowPrivateWebhookTargets = b.Config.Env != "production" && os.Getenv("WEBHOOKS_ALLOW_PRIVATE") == "true" return core.New(core.Deps{ - DB: b.DB, Signer: b.Signer, Encrypter: b.Encrypter, Email: b.Email, Notifier: n, Push: b.Push, + DB: b.DB, Signer: b.Signer, Encrypter: b.Encrypter, Email: b.Email, Notifier: n, Push: b.Push, Exports: b.Exports, Billing: Billing(b.Config, b.Log), Cursors: httpx.Cursors{Key: key}, Log: b.Log, Issuer: b.Config.Issuer, PublicBaseURL: b.Config.PublicBaseURL, AuthorizeURL: b.Config.AuthorizeURL, DashboardURL: b.Config.DashboardURL, CORSOrigins: b.Config.CORSOrigins, diff --git a/internal/platform/wire/exports.go b/internal/platform/wire/exports.go new file mode 100644 index 0000000..d3606f9 --- /dev/null +++ b/internal/platform/wire/exports.go @@ -0,0 +1,78 @@ +package wire + +import ( + "context" + "io" + "os" + "path/filepath" + "time" + + "github.com/aws/aws-sdk-go-v2/service/s3" + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" + + "github.com/CompositeCode/substratalapps.com/internal/core" + "github.com/CompositeCode/substratalapps.com/internal/platform/config" +) + +// Exports is where finished Audit Event exports go: the AUDIT_EXPORT_BUCKET +// S3 bucket in AWS, or .dev/audit-exports locally. +func Exports(ctx context.Context, cfg config.Config) (core.ExportStore, error) { + if cfg.AuditExportBucket == "" { + return LocalExports{Dir: filepath.Join(".dev", "audit-exports")}, nil + } + awsCfg, err := AWS(ctx) + if err != nil { + return nil, err + } + client := s3.NewFromConfig(awsCfg) + return S3Exports{Client: client, Presign: s3.NewPresignClient(client), Bucket: cfg.AuditExportBucket}, nil +} + +// S3Exports stores exports in a private bucket and hands out presigned GETs. +type S3Exports struct { + Client *s3.Client + Presign *s3.PresignClient + Bucket string +} + +func (e S3Exports) Put(ctx context.Context, key, contentType string, body io.ReadSeeker) error { + _, err := e.Client.PutObject(ctx, &s3.PutObjectInput{Bucket: &e.Bucket, Key: &key, Body: body, + ContentType: &contentType, ServerSideEncryption: s3types.ServerSideEncryptionAes256}) + return err +} + +func (e S3Exports) URL(ctx context.Context, key string, ttl time.Duration) (string, error) { + req, err := e.Presign.PresignGetObject(ctx, &s3.GetObjectInput{Bucket: &e.Bucket, Key: &key}, + s3.WithPresignExpires(ttl)) + if err != nil { + return "", err + } + return req.URL, nil +} + +// LocalExports writes exports under Dir, for local development. +type LocalExports struct{ Dir string } + +func (e LocalExports) Put(_ context.Context, key, _ string, body io.ReadSeeker) error { + path := filepath.Join(e.Dir, filepath.FromSlash(key)) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + f, err := os.Create(path) + if err != nil { + return err + } + if _, err := io.Copy(f, body); err != nil { + _ = f.Close() + return err + } + return f.Close() +} + +func (e LocalExports) URL(_ context.Context, key string, _ time.Duration) (string, error) { + abs, err := filepath.Abs(filepath.Join(e.Dir, filepath.FromSlash(key))) + if err != nil { + return "", err + } + return "file://" + filepath.ToSlash(abs), nil +} diff --git a/internal/testenv/testenv.go b/internal/testenv/testenv.go index 6d326b1..45d1c04 100644 --- a/internal/testenv/testenv.go +++ b/internal/testenv/testenv.go @@ -59,6 +59,7 @@ type Env struct { Mail *email.Capture Deliverer *webhooks.Deliverer Push *push.Recorder // every push the API sends, instead of APNs/FCM/Web Push + Exports *Exports // finished audit exports, instead of S3 Signer *auth.LocalSigner Admin string // superadmin access token AdminID string @@ -223,12 +224,14 @@ func NewWith(t testing.TB, opts Options) *Env { deliverer := webhooks.New(d, enc, mail, log) deliverer.AllowPrivate = true recorder := &push.Recorder{} - s := core.New(core.Deps{DB: d, Signer: signer(), Encrypter: enc, Email: mail, Push: recorder, + exports := &Exports{} + s := core.New(core.Deps{DB: d, Signer: signer(), Encrypter: enc, Email: mail, Push: recorder, Exports: exports, Cursors: httpx.Cursors{Key: []byte("test-cursor-key")}, Log: log, Issuer: "https://api.substratalapps.com", PublicBaseURL: "http://localhost", AuthorizeURL: "https://auth.substratalapps.com/authorize", DashboardURL: "https://dashboard.test", DisableRateLimits: !opts.RateLimits, Billing: opts.Billing}) srv := httptest.NewServer(s.Handler()) - e := &Env{T: t, Server: srv, Core: s, DB: d, Mail: mail, Deliverer: deliverer, Push: recorder, Signer: signer(), Validate: true, + e := &Env{T: t, Server: srv, Core: s, DB: d, Mail: mail, Deliverer: deliverer, Push: recorder, Exports: exports, + Signer: signer(), Validate: true, validator: specValidator(t)} t.Cleanup(func() { srv.Close(); d.Close() }) adminEmail := "admin+" + Unique() + "@example.com" @@ -241,6 +244,43 @@ func NewWith(t testing.TB, opts Options) *Env { return e } +// Exports is an in-memory core.ExportStore. Its URLs are +// https://exports.test/; Download reads one back. +type Exports struct { + mu sync.Mutex + files map[string][]byte +} + +func (x *Exports) Put(_ context.Context, key, _ string, body io.ReadSeeker) error { + b, err := io.ReadAll(body) + if err != nil { + return err + } + x.mu.Lock() + defer x.mu.Unlock() + if x.files == nil { + x.files = map[string][]byte{} + } + x.files[key] = b + return nil +} + +func (x *Exports) URL(_ context.Context, key string, ttl time.Duration) (string, error) { + return "https://exports.test/" + key + "?ttl=" + ttl.String(), nil +} + +// Download returns the file a URL from this store points at. +func (x *Exports) Download(rawURL string) ([]byte, bool) { + u, err := url.Parse(rawURL) + if err != nil { + return nil, false + } + x.mu.Lock() + defer x.mu.Unlock() + b, ok := x.files[strings.TrimPrefix(u.Path, "/")] + return b, ok +} + // Unique returns a short random suffix for emails, slugs, and names. func Unique() string { b := make([]byte, 5) diff --git a/migrations/0007_audit_exports.sql b/migrations/0007_audit_exports.sql new file mode 100644 index 0000000..e804b3d --- /dev/null +++ b/migrations/0007_audit_exports.sql @@ -0,0 +1,54 @@ +-- 0007_audit_exports: asynchronous Audit Event exports +-- (API Reference → Audit → Exporting). +-- +-- A request stores the caller's forced scope and filters; the jobs worker +-- writes the file to private storage and marks it ready. Rows and files +-- are deleted 7 days after the request. +-- +-- Numbered after 0005_test_user_expiry and 0006_self_registered_apps, +-- which ship first; the action list below keeps 0005's user.expired. + +create table audit_exports ( + id text primary key, + requested_by_type text not null check (requested_by_type in ('user','api_key')), + requested_by_id text not null, + scope jsonb not null, -- the forced scope: {application_id | target_user_id | tenant_id}, or {} + filters jsonb not null, + format text not null check (format in ('ndjson','csv')), + status text not null default 'pending' check (status in ('pending','running','ready','failed')), + row_count integer, + object_key text, + error text, + test_mode boolean not null default false, + created_at timestamptz not null default now(), + started_at timestamptz, + completed_at timestamptz, + expires_at timestamptz not null, + check ((status = 'ready') = (object_key is not null and row_count is not null)), + check ((status = 'failed') = (error is not null)) +); +create index audit_exports_due on audit_exports (created_at) where status in ('pending','running'); + +alter table audit_exports enable row level security; +alter table audit_exports force row level security; +create policy audit_exports_mode on audit_exports + using (substratal_util.mode_visible(test_mode)) with check (substratal_util.mode_visible(test_mode)); + +alter table audit_events drop constraint audit_events_action_check; +alter table audit_events add constraint audit_events_action_check check (action in ( + 'user.created','user.updated','user.email_changed','user.suspended','user.reactivated', + 'user.deleted','user.expired','user.erasure_requested','user.erasure_cancelled','user.erased', + 'user.password_changed','user.password_reset','user.mfa_enabled','user.mfa_disabled','user.mfa_reset', + 'entitlement.granted','entitlement.disabled','entitlement.revoked','entitlement.expired', + 'entitlement.updated','entitlement.member_scope_changed','entitlement.deleted', + 'role.created','role.updated','role.deleted','role.assigned','role.removed', + 'profile.updated','settings.updated', + 'application.created','application.updated','application.review_status_changed', + 'organization.created','organization.updated','organization.member_invited','organization.member_added', + 'organization.member_removed','organization.member_role_changed', + 'subscription.created','subscription.status_changed','subscription.members_changed', + 'tenant.tier_change_requested','tenant.tier_change_request_updated','tenant.tier_changed', + 'api_key.created','api_key.updated','api_key.rotated','api_key.revoked','api_key.restrict_destructive_disabled', + 'webhook.created','webhook.updated','webhook.deleted','webhook.secret_rotated', + 'push_credential.created','push_credential.updated','push_credential.deleted','device.deleted', + 'audit.exported')) not valid; diff --git a/plan-remaining.md b/plan-remaining.md index 2477dd4..5c9753f 100644 --- a/plan-remaining.md +++ b/plan-remaining.md @@ -58,7 +58,6 @@ Matches [Roadmap → Phase 3](https://compositecode.github.io/substratalapps.com - **SSO broker integration** (`sso_connections` becoming real, an actual WorkOS-or-equivalent wiring behind `POST /v1/auth/sso/{provider}/callback`) — the schema and the route exist from Phase 1 per [Auth → Native auth and per-Organization SSO](https://compositecode.github.io/substratalapps.com/api-reference/auth/#native-auth-and-per-organization-sso), deliberately built in advance precisely so this phase is additive wiring, not a migration. Triggered by a real Organization actually asking for it, not a phase-number default. - Self-service Application registration + review queue. The *process* is already fully specified — reviewer model (no dedicated assignment, any `applications.manage` holder works the FIFO queue), a 5-business-day policy SLA, the `rejected` status with required `review_notes` and edit-to-resubmit, non-cascading suspension — see [Applications → The review lifecycle](https://compositecode.github.io/substratalapps.com/domain-model/applications/#the-review-lifecycle) and [API Reference → Applications → Reviewing a submission](https://compositecode.github.io/substratalapps.com/api-reference/applications/#reviewing-a-submission). What's actually deferred to this phase is just the self-service *submission* endpoint itself (`POST /v1/applications` opening up beyond platform-admin-only) — the review mechanics it feeds into are build-ready today. - SCIM-style provisioning. -- Richer audit/compliance views (filtering, export) — driven by actual [Compliance](https://compositecode.github.io/substratalapps.com/compliance/) needs (a SOC 2 audit, a customer's data-export request), not built ahead of either. - **The hosted sign-in page** above, if no first-party app has needed it sooner. ## Dedicated tenancy infrastructure