diff --git a/cmd/devserver/main.go b/cmd/devserver/main.go index 0fe471e..93f4648 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 fb42950..5ca8355 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` | `/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` | `/audit-events/{id}` | same as above, for that event | Fetch one event. | +| `POST` | `/audit-exports` | same as `GET /audit-events`, for the filters sent | Export matching events to a file. | +| `GET` | `/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 /audit-events` @@ -79,6 +81,40 @@ GET /audit-events?target_user_id=usr_01JAG3Z9X8QS3F6K2M4N5P6R7S&since=2026-09-01 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 /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 /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. A CSV cell that would start with `=`, `+`, `-`, `@`, a tab, or a carriage return is prefixed with `'`, so a spreadsheet never runs it as a formula (`request_id` is whatever the caller sent as `X-Request-Id`). +- **Asynchronous.** The request returns `202` at once, and a job writes the file, usually within a minute. Poll `GET /audit-exports/{id}` until `status` is `ready` (or `failed`, with `error` saying why: `too_many_rows` over 200,000 events or `too_large` over 400 MB, so narrow the filters; `internal_error` if writing the file failed, so request it again). +- **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 b361d69..314bfe2 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -12,6 +12,16 @@ What's changed in this specification over time. Every decision is written into t --- +## 2026-10-22 — 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.8`, with 123 operations. + +- **`POST /audit-exports`** takes the same filters as `GET /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 /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 `0011_audit_exports`. Infrastructure: a private exports bucket with a 7-day lifecycle rule, and an `audit-exports` schedule every minute. + ## 2026-10-22 — Self-service Application registration `POST /applications` is open to developers, not just staff. See [API Reference → Applications](../api-reference/applications/#post-applications) and [The review lifecycle](../domain-model/applications/#the-review-lifecycle). `openapi.yaml` is now at `1.0.0-draft.7`. diff --git a/docs/domain-model/database-schema.md b/docs/domain-model/database-schema.md index 83ac4a2..50bddcf 100644 --- a/docs/domain-model/database-schema.md +++ b/docs/domain-model/database-schema.md @@ -439,6 +439,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. + ## scim_tokens | Column | Type | Constraint | diff --git a/docs/domain-model/orders-and-audit.md b/docs/domain-model/orders-and-audit.md index 388f2bc..0d285ea 100644 --- a/docs/domain-model/orders-and-audit.md +++ b/docs/domain-model/orders-and-audit.md @@ -124,6 +124,7 @@ Every value `action` can take. It's mirrored exactly by the `AuditAction` enum i | `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-usersiddevicesdeviceid) — **admin-only** | `device` | | `organization.scim_token_created` / `organization.scim_token_revoked` | An Organization's [SCIM](../../api-reference/scim/) token is created (or rotated), or revoked | `organization` | +| `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 bb7acb0..46f3982 100644 --- a/docs/non-functional-requirements.md +++ b/docs/non-functional-requirements.md @@ -72,6 +72,7 @@ Per-API-key token bucket, returning `429` with a `Retry-After` header on exhaust | `GET /users/{id}/export` | 5 per user per day | | `POST /organizations` | 10 per user per day | | `POST /applications`, self-service | 10 per user per day (staff with `applications.manage` aren't limited beyond the default) | +| `POST /audit-exports` | 20 per API Key or User per day | | `POST /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 /users/{id}/devices/{deviceId}/test` | 10 per device per hour | diff --git a/docs/openapi.yaml b/docs/openapi.yaml index be323c7..2c777ce 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.7" + version: "1.0.0-draft.8" summary: >- User accounts, auth, roles, per-app entitlements (the on/off access switch), profiles, settings, organizations, tenancy, subscription billing, @@ -2244,6 +2244,56 @@ 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 /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 /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/ApiVersion" + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/AuditExportRequest" } + responses: + "400": { $ref: "#/components/responses/BadRequest" } + "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 + parameters: + - $ref: "#/components/parameters/ApiVersion" + responses: + "400": { $ref: "#/components/responses/BadRequest" } + "200": + description: The export + content: + application/json: + schema: { $ref: "#/components/schemas/AuditExport" } + "404": { $ref: "#/components/responses/NotFound" } + # ───────────────────────────── Webhooks ───────────────────────────── /webhooks: @@ -3799,6 +3849,7 @@ components: - user.mfa_enabled - user.mfa_disabled - user.mfa_reset + - audit.exported - entitlement.granted - entitlement.disabled - entitlement.revoked @@ -3846,7 +3897,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 /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: too_many_rows (over 200,000), too_large (over 400 MB), or internal_error." } + 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 } ScimToken: type: object diff --git a/docs/openapi/index.md b/docs/openapi/index.md index 92451e6..8e802d8 100644 --- a/docs/openapi/index.md +++ b/docs/openapi/index.md @@ -28,7 +28,7 @@ The complete Platform API contract as an [OpenAPI 3.1](https://spec.openapis.org | API version | `Substratal-Version: 1` request header, required (see [Conventions → Versioning](../api-reference/conventions/#versioning)) | | Authentication | `Authorization: Bearer `: a user access token or an API Key (`bearerAuth`) | | Content type | `application/json; charset=utf-8` | -| Operations | 121, every one with a stable `operationId` | +| Operations | 123, 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). @@ -55,10 +55,10 @@ Operations are grouped by tag, and each tag matches an API Reference page that e | Applications | 4 | [Applications](../api-reference/applications/) | | Entitlements | 6 | [Entitlements](../api-reference/entitlements/) | | Roles & Permissions | 10 | [Roles & Permissions](../api-reference/roles-and-permissions/) | -| Organizations | 11 | [Organizations](../api-reference/organizations/) | +| Organizations | 14 | [Organizations](../api-reference/organizations/) | | Tenancy | 6 | [Tenancy](../api-reference/tenancy/) | | Billing | 7 | [Billing](../api-reference/billing/) | -| Audit | 2 | [Audit](../api-reference/audit/) | +| Audit | 4 | [Audit](../api-reference/audit/) | | Webhooks | 9 | [Webhooks](../api-reference/webhooks/) | | Notifications | 17 | [Push Notifications](../api-reference/notifications/) | | API Keys | 6 | [API Keys](../api-reference/api-keys/) | @@ -67,6 +67,7 @@ Some endpoints aren't in the document's `paths`, because they follow external st - `GET /.well-known/jwks.json` and `GET /.well-known/openid-configuration`: signing keys and discovery, for verifying app tokens with any standard JWT library. See [Auth → Signing keys](../api-reference/auth/#signing-keys-jwks). - `/mcp`: the [MCP Server](../mcp-server/), generated from this document's `operationId`s. +- `/scim/v2/…`: [SCIM 2.0](../api-reference/scim/) provisioning, following RFC 7644. ## Authentication in the spec diff --git a/docs/roadmap.md b/docs/roadmap.md index 724f7d8..8f5fe2f 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 /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. **Built 2026-10-21:** [SCIM 2.0 Users](../api-reference/scim/), as invitations and membership removal. - 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. **Built 2026-10-21.** Self-registered Applications still need the hosted sign-in page above before they can go live. See [App developer/publisher model](../domain-model/applications/#the-review-lifecycle). diff --git a/infra/README.md b/infra/README.md index c6c4084..19ac8e7 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 /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 83a671b..39bad74 100644 --- a/infra/terraform/prod/main.tf +++ b/infra/terraform/prod/main.tf @@ -157,6 +157,38 @@ resource "aws_s3_bucket_versioning" "audit_archive" { versioning_configuration { status = "Enabled" } } +# Finished Audit Event exports (POST /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" { @@ -244,9 +276,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}/*" }]) } @@ -315,11 +351,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}/*"] }]) } @@ -452,6 +491,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 573009d..07481e2 100644 --- a/internal/api/gen/gen.go +++ b/internal/api/gen/gen.go @@ -356,6 +356,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" @@ -428,6 +429,8 @@ func (e AuditAction) Valid() bool { return true case AuditActionApplicationUpdated: return true + case AuditActionAuditExported: + return true case AuditActionDeviceDeleted: return true case AuditActionEntitlementDeleted: @@ -535,10 +538,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" @@ -559,6 +644,8 @@ func (e AuditTargetType) Valid() bool { return true case AuditTargetTypeApplication: return true + case AuditTargetTypeAuditExport: + return true case AuditTargetTypeDevice: return true case AuditTargetTypeEntitlement: @@ -2441,6 +2528,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: too_many_rows (over 200,000), too_large (over 400 MB), or internal_error. + 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 /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 @@ -3627,6 +3770,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"` @@ -4262,6 +4411,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 @@ -4781,6 +4933,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) @@ -5994,6 +6152,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) { @@ -10353,6 +10582,8 @@ func HandlerWithOptions(si ServerInterface, options StdHTTPServerOptions) http.H m.HandleFunc(http.MethodPatch+" "+options.BaseURL+"/applications/{id}/push-credentials/{provider}", wrapper.PushCredentialsUpdate) 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.MethodPost+" "+options.BaseURL+"/auth/app-tokens", wrapper.AuthCreateAppToken) m.HandleFunc(http.MethodPost+" "+options.BaseURL+"/auth/email/verify", wrapper.AuthVerifyEmail) m.HandleFunc(http.MethodPost+" "+options.BaseURL+"/auth/email/verify/resend", wrapper.AuthResendEmailVerification) diff --git a/internal/core/audit.go b/internal/core/audit.go index 8b64558..608c652 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 /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..b9b5405 --- /dev/null +++ b/internal/core/audit_exports.go @@ -0,0 +1,390 @@ +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 + // The jobs Lambda writes the file to its 512 MB /tmp before uploading, + // so both caps keep a large export from filling it. + auditExportMaxRows = 200_000 + auditExportMaxBytes = 400 << 20 + // The worker stops claiming new exports after this, well inside the + // jobs Lambda's 15-minute timeout; the next run picks them up. + auditExportBudget = 10 * time.Minute + 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 /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 /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 + } + start := time.Now() + err = eachMode(func(tm bool) error { + for time.Since(start) < auditExportBudget { + 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 + } + // A failure ends that export as failed rather than leaving it + // running to be retried forever; the caller can request again. + if err := s.runAuditExport(ctx, tm, claimed[0]); err != nil { + s.Log.Error("audit export failed", "export_id", claimed[0].ID, "err", err) + if err := s.failAuditExport(context.WithoutCancel(ctx), tm, claimed[0], "internal_error"); err != nil { + return err + } + } + done++ + } + return nil + }) + 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()) }() + + out := &countingWriter{w: file} + 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(out) + 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(out) + 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 out.n > auditExportMaxBytes { + return s.failAuditExport(ctx, tm, x, "too_large") + } + 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 { + row := []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)} + for i, v := range row { + row[i] = csvSafe(v) + } + return row +} + +// csvSafe keeps a cell from being read as a formula when the file is +// opened in a spreadsheet (CSV injection). request_id comes from the +// caller's X-Request-Id header, so a cell can start with anything. +func csvSafe(v string) string { + if v != "" && strings.ContainsRune("=+-@\t\r", rune(v[0])) { + return "'" + v + } + return v +} + +// countingWriter counts the bytes written through it. +type countingWriter struct { + w io.Writer + n int64 +} + +func (c *countingWriter) Write(p []byte) (int, error) { + n, err := c.w.Write(p) + c.n += int64(n) + return n, err +} + +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..8da97e8 --- /dev/null +++ b/internal/core/audit_exports_test.go @@ -0,0 +1,142 @@ +package core_test + +import ( + "bufio" + "bytes" + "context" + "encoding/csv" + "encoding/json" + "strings" + "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("/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("/users/"+u.ID+"/profile", e.Admin, map[string]any{"display_name": "Renamed"}).Expect(200) + want := e.Get("/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("/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("/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("/audit-exports", e.Admin, key, map[string]any{"format": "csv", "target_user_id": u.ID}).Expect(202).Str("id") + b := e.PostIdem("/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) + } + + // A cell that would start a spreadsheet formula is neutralized: + // request_id is whatever the caller sent as X-Request-Id. + e.Do(testenv.Req{Method: "PATCH", Path: "/users/" + u.ID + "/profile", Token: e.Admin, + Body: map[string]any{"display_name": "Again"}, Header: map[string]string{"X-Request-Id": "=HYPERLINK(\"x\")"}}).Expect(200) + inj := e.PostIdem("/audit-exports", e.Admin, testenv.Unique(), map[string]any{"format": "csv", "target_user_id": u.ID}).Expect(202).Str("id") + irows, err := csv.NewReader(bytes.NewReader(download(t, e, runExport(t, e, e.Admin, inj)))).ReadAll() + if err != nil { + t.Fatalf("csv: %v", err) + } + found := false + for _, r := range irows[1:] { + if strings.Contains(r[11], "HYPERLINK") { + found = true + if r[11] != `'=HYPERLINK("x")` { + t.Fatalf("request_id cell = %q", r[11]) + } + } + } + if !found { + t.Fatal("the planted request_id isn't in the export") + } + + // Validation. + e.PostIdem("/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("/audit-exports", u.Token, testenv.Unique(), map[string]any{}).ExpectErr(403, "forbidden") + own := e.PostIdem("/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("/audit-exports/"+id, u.Token).ExpectErr(404, "audit_export_not_found") + + // An app-confined key is forced to its own Application. + appKey := e.Post("/api-keys", e.Admin, map[string]any{"name": "auditor", "scope": app, + "permissions": []string{"audit.view"}}).Expect(201).Str("secret") + scoped := e.PostIdem("/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 124d3ae..6e5f6e9 100644 --- a/internal/core/server.go +++ b/internal/core/server.go @@ -67,6 +67,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 df71a73..f7d5062 100644 --- a/internal/core/system.go +++ b/internal/core/system.go @@ -402,6 +402,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 f69de27..9680cee 100644 --- a/internal/ids/ids.go +++ b/internal/ids/ids.go @@ -35,6 +35,7 @@ const ( Notification = "ntf_" NotificationDelivery = "ndl_" SCIMToken = "sct_" + AuditExport = "aex_" ) // Credential prefixes. Never logged, never echoed after issuance. diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index 42d013e..549a6f9 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -61,6 +61,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 1f25da5..d858f63 100644 --- a/internal/mcp/tools.json +++ b/internal/mcp/tools.json @@ -974,6 +974,172 @@ "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 /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 /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 /audit-exports", + "inputSchema": { + "additionalProperties": false, + "properties": { + "body": { + "description": "The same filters as GET /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.expired", + "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", + "organization.scim_token_created", + "organization.scim_token_revoked", + "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": "/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 +1174,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 /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": "/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)", @@ -1033,6 +1233,7 @@ "user.mfa_enabled", "user.mfa_disabled", "user.mfa_reset", + "audit.exported", "entitlement.granted", "entitlement.disabled", "entitlement.revoked", @@ -1129,7 +1330,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 56016f7..f5209bb 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 95c8774..3e381eb 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 @@ -269,12 +270,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, Now: opts.Now}) 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" @@ -287,6 +290,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/0011_audit_exports.sql b/migrations/0011_audit_exports.sql new file mode 100644 index 0000000..bfb6921 --- /dev/null +++ b/migrations/0011_audit_exports.sql @@ -0,0 +1,56 @@ +-- 0011_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 0008_scim_tokens (in production), 0009_test_user_expiry, +-- and 0010_self_registered_apps. The audit action list is the full set, +-- so whichever order those land in, no action is dropped. + +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', + 'organization.scim_token_created','organization.scim_token_revoked', + '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 77fa6f9..c0b5f9b 100644 --- a/plan-remaining.md +++ b/plan-remaining.md @@ -40,7 +40,6 @@ The API is complete, but the user-facing pages its emails and redirects point to Matches [Roadmap → Phase 3](https://compositecode.github.io/substratalapps.com/roadmap/#phase-3) — explicitly **not** built speculatively ahead of the triggers named in [Scale-out](#scale-out) and [Applications → The review lifecycle](https://compositecode.github.io/substratalapps.com/domain-model/applications/#the-review-lifecycle): - **SSO broker integration** (`sso_connections` becoming real, an actual WorkOS-or-equivalent wiring behind `POST /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. -- 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. Self-service Application registration is built, so this is now what stands between a self-registered Application and going live. ## Dedicated tenancy infrastructure