diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 15d85f5a..feaf0fca 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -5,6 +5,12 @@ on: branches: [main] pull_request: branches: [main] + # Weekly schedule keeps the vuln-DB scans honest even on quiet weeks + # where no PRs land. Sunday 06:00 UTC = light load on GitHub runners. + # The security-scan job below gates on github.event_name to stay + # cheap on the non-security jobs. + schedule: + - cron: '0 6 * * 0' # Stop in-flight runs when a new commit pushes to the same PR. concurrency: @@ -287,10 +293,125 @@ jobs: --next ../../apps/admin/.next \ --summary "$GITHUB_STEP_SUMMARY" + # Security scanners — issues #190 and #196. + # + # Five tools, grouped into one job so a green check covers them all + # for branch protection: + # + # - govulncheck (#190): Go vuln DB, looks at imports + reachability. + # - osv-scanner (#190): Google's cross-language vuln DB (Go, npm, etc). + # - semgrep (#190): rules-based static analysis. OWASP Top 10 + + # security-audit registry packs. + # - gosec (#196): Go-specific security linter (G-rules). + # - gitleaks (#196): secret scanning across the working tree. + # + # Fail-fast: any one of these returning non-zero fails the job. The + # scanners are NOT advisory — a known CVE landing in go.mod should + # block the merge. (The acceptable-CVE list lives in osv-scanner's + # config + .semgrepignore + .gitleaks.toml; raise findings via PR.) + # + # Runs on PR + the weekly schedule above. Pull requests get the same + # full sweep as the cron run so a freshly-disclosed CVE in main + # doesn't sit unaddressed until the next Sunday. + security-scan: + runs-on: ubuntu-latest + permissions: + contents: read + security-events: write + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 # gitleaks needs history for the full sweep + - uses: actions/setup-go@v5 + with: + go-version-file: 'go.work' + cache: true + - name: go work sync + run: go work sync + + # govulncheck — Go's first-party vulnerability scanner. Walks the + # workspace modules; reports CVEs that are reachable from at least + # one entrypoint. We iterate the workspace `use` list (same + # pattern as lint-go) so transitive non-workspace modules don't + # noisy up the report. + - name: Install govulncheck + run: go install golang.org/x/vuln/cmd/govulncheck@latest + - name: govulncheck (workspace modules) + run: | + for dir in $(go work edit -json | python3 -c "import json,sys; [print(u['DiskPath']) for u in json.load(sys.stdin)['Use']]"); do + echo "::group::govulncheck $dir" + (cd "$dir" && govulncheck ./...) + echo "::endgroup::" + done + + # osv-scanner — cross-language vuln DB (covers npm + Go + container + # images). Runs from the repo root with -r so it finds every + # manifest (go.mod, pnpm-lock.yaml, package.json, ...). + - name: Install osv-scanner + run: go install github.com/google/osv-scanner/cmd/osv-scanner@latest + - name: osv-scanner (recursive) + run: osv-scanner -r . + + # semgrep — rules-based SAST. OWASP Top 10 catches the obvious + # injection / authz patterns; security-audit is a broader sweep + # of misuse patterns Semgrep has community rules for. + - name: Install semgrep + run: | + python3 -m pip install --quiet --upgrade pip + python3 -m pip install --quiet semgrep + - name: semgrep (OWASP + security-audit) + run: | + semgrep \ + --config=p/owasp-top-ten \ + --config=p/security-audit \ + --error \ + --metrics=off \ + ./apps ./packages + + # gosec — Go-specific security linter (G101 hardcoded creds, G104 + # ignored errors on security calls, G304 file inclusion, etc.). + # Iterates the workspace `use` list like govulncheck for the + # same reason. + - name: Install gosec + run: go install github.com/securego/gosec/v2/cmd/gosec@latest + - name: gosec (workspace modules) + run: | + for dir in $(go work edit -json | python3 -c "import json,sys; [print(u['DiskPath']) for u in json.load(sys.stdin)['Use']]"); do + echo "::group::gosec $dir" + (cd "$dir" && gosec ./...) + echo "::endgroup::" + done + + # gitleaks — secret scanning. --no-banner keeps the log tight; + # --redact strips matched secrets from the report (we never want + # a leaked token mirrored into a public CI log); --report-format + # json so a follow-up step (or human reviewer) can ingest the + # findings cleanly. The non-zero exit on findings is the gate. + - name: Install gitleaks + run: | + GITLEAKS_VERSION=8.21.2 + curl -sSfL "https://github.com/gitleaks/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_linux_x64.tar.gz" \ + | tar -xz -C /tmp gitleaks + sudo mv /tmp/gitleaks /usr/local/bin/gitleaks + - name: gitleaks detect + run: | + gitleaks detect \ + --no-banner \ + --redact \ + --report-format json \ + --report-path gitleaks-report.json + - name: Upload gitleaks report + if: always() + uses: actions/upload-artifact@v4 + with: + name: gitleaks-report + path: gitleaks-report.json + if-no-files-found: ignore + # Aggregate gate. Required check on the branch protection rule. # bundle-budget is intentionally NOT in `needs:` while it remains advisory. ci: - needs: [lint-docs, lint-go, test-go, lint-web, test-web, lint-dashboards, lint-openapi] + needs: [lint-docs, lint-go, test-go, lint-web, test-web, lint-dashboards, lint-openapi, security-scan] if: always() runs-on: ubuntu-latest steps: diff --git a/.github/workflows/sbom.yml b/.github/workflows/sbom.yml new file mode 100644 index 00000000..4cef00f3 --- /dev/null +++ b/.github/workflows/sbom.yml @@ -0,0 +1,112 @@ +name: SBOM + +# Software Bill of Materials generation (issue #142). +# +# Triggers: +# - tag push (v*): publish CycloneDX SBOMs as release artifacts. +# - pull_request: upload SBOMs as workflow artifacts only (no release). +# - workflow_dispatch: manual dry-run from the Actions tab. +# +# Scope (one SBOM per shipped surface): +# - apps/api — Go module (cmd/server binary) +# - apps/web — Next.js bundle (transitive npm tree) +# - cli/gonext — Go module (gonext CLI binary) +# - examples/plugins/seo — WASM plugin (.gnplugin) Go module +# +# Tool choice: anchore/syft. CycloneDX 1.5 JSON is the lingua franca our +# downstream SOC2 evidence pipeline expects. We do NOT use SPDX here +# because the dependency-track instance the security team runs ingests +# CycloneDX natively; pivoting later is one --output flag away. + +on: + push: + tags: + - 'v*' + pull_request: + branches: [main] + workflow_dispatch: + +permissions: + contents: write # required to attach SBOMs to the GitHub Release on tag push + pull-requests: read + +concurrency: + group: sbom-${{ github.ref }} + cancel-in-progress: true + +jobs: + sbom: + name: Generate SBOM (${{ matrix.target.name }}) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + target: + - name: api + path: apps/api + kind: go + - name: web + path: apps/web + kind: npm + - name: cli-gonext + path: cli/gonext + kind: go + - name: plugin-seo + path: examples/plugins/seo + kind: go + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 1 + + # Syft is fetched as a static binary instead of relying on the + # action's docker image — this keeps the job cold-start under 15s + # and avoids the action's implicit privileges on the workflow. + - name: Install syft + run: | + curl -sSfL https://raw.githubusercontent.com/anchore/syft/main/install.sh \ + | sh -s -- -b /usr/local/bin v1.18.1 + syft version + + - name: Generate CycloneDX SBOM + run: | + set -euo pipefail + mkdir -p sbom-out + syft scan "dir:${{ matrix.target.path }}" \ + -o cyclonedx-json="sbom-out/sbom-${{ matrix.target.name }}.cdx.json" \ + -o spdx-json="sbom-out/sbom-${{ matrix.target.name }}.spdx.json" + echo "::group::SBOM summary" + jq '{component: .metadata.component.name, components: (.components|length)}' \ + "sbom-out/sbom-${{ matrix.target.name }}.cdx.json" + echo "::endgroup::" + + - name: Upload SBOM artifact + uses: actions/upload-artifact@v4 + with: + name: sbom-${{ matrix.target.name }} + path: sbom-out/ + retention-days: 30 + if-no-files-found: error + + # On tag push, attach every SBOM to the GitHub Release. We collect the + # matrix outputs into a single release-upload step so the operator sees + # one job in the release timeline rather than four. + release: + name: Attach SBOMs to release + needs: sbom + if: startsWith(github.ref, 'refs/tags/v') + runs-on: ubuntu-latest + steps: + - name: Download all SBOMs + uses: actions/download-artifact@v4 + with: + pattern: sbom-* + path: sbom-out + merge-multiple: true + + - name: Attach SBOMs to release + uses: softprops/action-gh-release@v2 + with: + files: sbom-out/*.json + fail_on_unmatched_files: true + generate_release_notes: false diff --git a/.github/workflows/zap-dast.yml b/.github/workflows/zap-dast.yml new file mode 100644 index 00000000..ec78e664 --- /dev/null +++ b/.github/workflows/zap-dast.yml @@ -0,0 +1,127 @@ +name: ZAP DAST + +# OWASP ZAP baseline scan against the running compose stack (issue #204). +# +# Triggers: +# - cron nightly 03:00 UTC (off-peak so the scan never collides with +# the daily release window) +# - workflow_dispatch (manual runs from the Actions tab) +# +# We intentionally do NOT trigger on pull_request: ZAP runs take ~5 +# minutes and the false-positive rate on a CMS API would noise the PR +# checkers. The nightly cadence matches the security team's review SLO. +# +# Pipeline: +# 1. Boot the dev compose stack via `make up`. +# 2. Poll /readyz until the API is healthy (compose --wait covers the +# depends_on healthchecks, but we re-confirm against the public +# readiness endpoint). +# 3. Run zap-baseline.py against http://api:8080 from inside the +# compose network — the ZAP container joins the same network so +# DNS for "api" resolves. +# 4. Upload the HTML + JSON reports as a workflow artifact. +# 5. Tear down the stack (always, even on failure). +# +# Why we don't fail the job on findings: DAST has a high false-positive +# rate, especially against a CMS that ships permissive media endpoints +# and a versioned plugin sandbox. The triage happens off-line against +# the uploaded artifact; the workflow's job is to KEEP RUNNING the scan, +# not to block deploys on it. The security team owns weekly triage. + +on: + schedule: + - cron: '0 3 * * *' + workflow_dispatch: + inputs: + target_url: + description: Target URL to scan (defaults to the in-network API). + required: false + default: 'http://api:8080' + +permissions: + contents: read + issues: write # ZAP action opens an issue when run on a public repo; we keep the permission narrow. + +concurrency: + group: zap-dast + cancel-in-progress: false + +jobs: + baseline: + name: ZAP baseline scan + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + + - name: Boot compose stack + run: | + set -euo pipefail + make up + docker compose ps + + # The compose 'depends_on: condition: service_healthy' fence is + # the source of truth, but we still wait against /readyz so the + # job log shows a clean ready line before ZAP fires. + - name: Wait for API /readyz + run: | + set -euo pipefail + for i in $(seq 1 60); do + if curl -fsS http://localhost:8080/readyz >/dev/null 2>&1; then + echo "API ready after ${i}s" + exit 0 + fi + sleep 1 + done + echo "API never became ready" >&2 + docker compose logs api | tail -100 >&2 + exit 1 + + - name: Create ZAP working dir + run: | + mkdir -p zap-out + chmod 777 zap-out + + # We run ZAP inside the compose network so it can reach "api" by + # service name and so its outbound HTTP traffic stays off the + # host's localhost. zap-baseline.py is the recommended entrypoint + # for "passive scan + safe active checks" — full active scans live + # in a separate, hand-triggered workflow. + - name: Run ZAP baseline + run: | + set +e + docker run --rm \ + --network "${COMPOSE_PROJECT_NAME:-gonext}_default" \ + -v "${{ github.workspace }}/zap-out:/zap/wrk:rw" \ + ghcr.io/zaproxy/zaproxy:stable \ + zap-baseline.py \ + -t "${{ github.event.inputs.target_url || 'http://api:8080' }}" \ + -r zap-report.html \ + -J zap-report.json \ + -m 5 \ + -T 10 + zap_status=$? + # zap-baseline returns: + # 0 = no warnings, 1 = warnings, 2 = false positives only, + # 3 = scan failed. We surface a non-zero only on 3 because + # warnings are expected (see workflow header note). + if [ "$zap_status" -eq 3 ]; then + echo "ZAP scan itself failed (exit 3); failing job." + exit 3 + fi + echo "ZAP exit: $zap_status (treated as success — findings triaged offline)." + exit 0 + + - name: Upload ZAP report + if: always() + uses: actions/upload-artifact@v4 + with: + name: zap-report-${{ github.run_id }} + path: zap-out/ + retention-days: 30 + if-no-files-found: warn + + - name: Tear down compose stack + if: always() + run: | + docker compose down --volumes --remove-orphans || true diff --git a/apps/admin/src/app/(authenticated)/settings/privacy/components/PrivacyActions.tsx b/apps/admin/src/app/(authenticated)/settings/privacy/components/PrivacyActions.tsx new file mode 100644 index 00000000..7ba12661 --- /dev/null +++ b/apps/admin/src/app/(authenticated)/settings/privacy/components/PrivacyActions.tsx @@ -0,0 +1,280 @@ +'use client'; + +/** + * PrivacyActions — the interactive surface of /settings/privacy. + * + * Two cards, one per action. The card surfaces are the paper-2 + * pattern shared across /settings; the destructive card carries a + * subtle red border so the operator never confuses it for the safe + * export action. + * + * State machine: + * + * exportState: + * idle → loading → success({ jobId, pollUrl }) + * \-> error(message) + * + * deleteState: + * idle → confirming → loading → done + * \-> error(message) + * + * The export does not poll on its own; the operator follows the + * polling URL surfaced in the success banner. That URL is wired by a + * follow-up issue and lives at /api/v1/account/data/export/{jobId}. + */ + +import type { FormEvent, ReactElement } from 'react'; +import { useState } from 'react'; +import { AlertTriangle, Download, Loader2, Trash2 } from 'lucide-react'; + +import { ApiError, api } from '@/lib/api-client'; +import { Button } from '@/components/ui/button'; + +interface ExportSuccess { + jobId: string; + pollUrl: string; +} + +type ExportState = + | { kind: 'idle' } + | { kind: 'loading' } + | { kind: 'success'; data: ExportSuccess } + | { kind: 'error'; message: string }; + +type DeleteState = + | { kind: 'idle' } + | { kind: 'confirming' } + | { kind: 'loading' } + | { kind: 'done' } + | { kind: 'error'; message: string }; + +interface ExportResponse { + job_id: string; + status: string; + poll_url: string; + created_at: string; +} + +interface DeleteResponse { + anonymized_at: string; + scheduled_purge_at: string; +} + +function describeError(err: unknown): string { + if (err instanceof ApiError) { + return err.message || `Request failed with HTTP ${err.status}.`; + } + if (err instanceof Error) return err.message; + return 'Unknown error.'; +} + +export function PrivacyActions(): ReactElement { + const [exportState, setExportState] = useState({ kind: 'idle' }); + const [deleteState, setDeleteState] = useState({ kind: 'idle' }); + const [password, setPassword] = useState(''); + const [passwordConfirm, setPasswordConfirm] = useState(''); + + async function handleExport(): Promise { + setExportState({ kind: 'loading' }); + try { + const resp = await api.get('/api/v1/account/data/export'); + setExportState({ + kind: 'success', + data: { jobId: resp.job_id, pollUrl: resp.poll_url }, + }); + } catch (err) { + setExportState({ kind: 'error', message: describeError(err) }); + } + } + + async function handleDelete(e: FormEvent): Promise { + e.preventDefault(); + if (password !== passwordConfirm) { + setDeleteState({ + kind: 'error', + message: 'Passwords do not match. Type the same password twice to confirm.', + }); + return; + } + setDeleteState({ kind: 'loading' }); + try { + await api.post('/api/v1/account/data/delete', { + password, + password_confirm: passwordConfirm, + }); + setDeleteState({ kind: 'done' }); + // Clear the password material from React state ASAP — the + // user agent's BFCache may keep the form alive after navigation. + setPassword(''); + setPasswordConfirm(''); + } catch (err) { + setDeleteState({ kind: 'error', message: describeError(err) }); + } + } + + return ( +
+ {/* --- Export card -------------------------------------------- */} +
+
+

+ Download your data +

+

+ Generates a ZIP containing your profile, posts you authored, + comments, uploaded media, and your audit-log rows. Limit: one + export per day per account. +

+
+ +
+ +
+ + {exportState.kind === 'success' && ( +
+

+ Export queued. Job id {exportState.data.jobId}. +

+

+ Check status:{' '} + + {exportState.data.pollUrl} + +

+
+ )} + + {exportState.kind === 'error' && ( +
+

Export failed: {exportState.message}

+
+ )} +
+ + {/* --- Delete card -------------------------------------------- */} +
+
+

+

+

+ Anonymises every record we hold for you in place: posts and + comments you authored become "Deleted User", + uploaded media is unattached, your profile is wiped. The + account is fully purged 30 days later. There is no undo + after the purge runs. +

+
+ + {deleteState.kind === 'done' ? ( +
+

+ Account anonymised. Your session will end shortly; the final + purge is scheduled for 30 days from now. +

+
+ ) : ( +
+ + + + {deleteState.kind === 'error' && ( +
+ {deleteState.message} +
+ )} + +
+ +
+
+ )} +
+
+ ); +} diff --git a/apps/admin/src/app/(authenticated)/settings/privacy/page.tsx b/apps/admin/src/app/(authenticated)/settings/privacy/page.tsx new file mode 100644 index 00000000..fa16d591 --- /dev/null +++ b/apps/admin/src/app/(authenticated)/settings/privacy/page.tsx @@ -0,0 +1,65 @@ +/** + * /settings/privacy — GDPR self-service surface (issue #216). + * + * Two actions, both irreversible to varying degrees: + * + * 1. Download your data — kicks off an async export job. The + * worker assembles a ZIP and returns a download URL through the + * poll endpoint; this page surfaces the job id and shows a + * banner with the polling URL. + * + * 2. Delete account — anonymises the user in place and schedules a + * hard-delete 30 days out. Requires the current password (typed + * twice) so an accidental click can't destroy data. After a + * successful delete the API also invalidates every session, so + * the next page navigation kicks the user back to the login + * screen. + * + * Styled against the Living-Systems brand: cream paper, Archivo + * headline with the italic accent, emerald CTA for export, red + * destructive CTA for delete. See docs/design/HANDOFF.md. + * + * The client component is deliberately small — the heavy lifting + * happens on the server. We do NOT pre-fetch any data on this page + * because both actions are write-only. + */ +import type { ReactElement } from 'react'; +import Link from 'next/link'; +import { ArrowLeft, ShieldCheck } from 'lucide-react'; + +import { Headline } from '@/components/ui/headline'; + +import { PrivacyActions } from './components/PrivacyActions'; + +export default function PrivacyPage(): ReactElement { + return ( +
+
+ +
+ + +
+ ); +} diff --git a/apps/api/cmd/server/main.go b/apps/api/cmd/server/main.go index 2471fe34..fbb6fbac 100644 --- a/apps/api/cmd/server/main.go +++ b/apps/api/cmd/server/main.go @@ -27,9 +27,11 @@ import ( "strings" "time" + "github.com/hibiken/asynq" "github.com/jackc/pgx/v5/pgxpool" goredis "github.com/redis/go-redis/v9" + accountdata "github.com/Singleton-Solution/GoNext/apps/api/internal/account/data" admincomments "github.com/Singleton-Solution/GoNext/apps/api/internal/admin/comments" adminmedia "github.com/Singleton-Solution/GoNext/apps/api/internal/admin/media" adminthemes "github.com/Singleton-Solution/GoNext/apps/api/internal/admin/themes" @@ -903,6 +905,57 @@ func buildRouter(cfg *config.Config, pool *pgxpool.Pool, rdb *goredis.Client, se } } + // GDPR account-data routes (issue #216): + // GET /api/v1/account/data/export — async export + // POST /api/v1/account/data/delete — anonymise + schedule purge + // + // Wired only when the pool, redis, and sessions are all healthy: + // the export handler enqueues to the worker via Asynq (needs + // Redis), the delete handler runs a multi-row UPDATE under a + // transaction (needs the pool), and both endpoints sit behind + // RequireSession. + if pool == nil || rdb == nil || sessions == nil { + logger.Warn("account/data: skipping mount; pool, redis, or sessions is nil") + } else { + // Build the Asynq client from the same Redis URL the worker + // consumes. We never share a *redis.Client with Asynq because + // the client manages its own pool and connection lifecycle — + // passing rdb directly would interleave Asynq's BRPOP traffic + // with the rest of the app's redis usage. + var asynqClient *asynq.Client + if cfg.Redis.URL != "" { + if redisOpt, err := asynq.ParseRedisURI(cfg.Redis.URL); err != nil { + logger.Warn("account/data: failed to parse redis URL for asynq", + slog.Any("err", err)) + } else { + asynqClient = asynq.NewClient(redisOpt) + // Note: this asynq.Client owns its own pool of Redis + // connections, separate from the rdb client. The + // process exit closes it implicitly; we don't register + // it with the shutdown orchestrator because that + // instance lives in main() not buildRouter. + } + } + if asynqClient == nil { + logger.Warn("account/data: skipping mount; asynq client not available") + } else { + dataHandlers := accountdata.NewHandlers(accountdata.Deps{ + Verifier: accountdata.NewPgxPasswordVerifier(pool, []byte(cfg.Auth.Pepper)), + Anonymizer: accountdata.NewPgxAnonymizer(pool), + Enqueuer: accountdata.NewAsynqEnqueuer(asynqClient, "default"), + Audit: auditEmitter, + Log: logger, + PollURLBase: strings.TrimRight(cfg.Email.SiteURL, "/"), + }) + guarded := authmw.RequireSession(sessions)(dataHandlers.Routes()) + mux.Handle("/api/v1/account/data/export", guarded) + mux.Handle("/api/v1/account/data/delete", guarded) + logger.Info("account/data: routes mounted", + slog.String("export", "/api/v1/account/data/export"), + slog.String("delete", "/api/v1/account/data/delete")) + } + } + // Posts REST surface (CRUD over /api/v1/posts). PgxStore persists // to the canonical posts table from 000004_posts.up.sql — restart // the api and the corpus is still there. The store is constructed diff --git a/apps/api/internal/account/data/asynq_enqueuer.go b/apps/api/internal/account/data/asynq_enqueuer.go new file mode 100644 index 00000000..4bd0df3b --- /dev/null +++ b/apps/api/internal/account/data/asynq_enqueuer.go @@ -0,0 +1,67 @@ +package data + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/hibiken/asynq" +) + +// asyncTaskName is the canonical task name the worker registers +// (see apps/worker/internal/tasks/gdpr.TaskExportRun). We keep the +// string literal here rather than import the worker package — that +// would create an apps-cross-apps dependency the build doesn't allow. +const asyncTaskName = "gdpr.export.run" + +// AsynqEnqueuer is the production [ExportEnqueuer] backed by Asynq's +// client. The caller owns the underlying client's lifecycle; we hold +// only a borrowed reference. +type AsynqEnqueuer struct { + client *asynq.Client + queue string +} + +// NewAsynqEnqueuer wraps a client. Queue defaults to "default" if +// empty — operators who want a dedicated GDPR queue (matching the +// "critical" queue used by the purge tick) pass it through. +func NewAsynqEnqueuer(client *asynq.Client, queue string) *AsynqEnqueuer { + if client == nil { + panic("data.NewAsynqEnqueuer: client is required") + } + if queue == "" { + queue = "default" + } + return &AsynqEnqueuer{client: client, queue: queue} +} + +// payload mirrors gdpr.ExportPayload (in apps/worker/internal/tasks/gdpr). +// Keeping a private copy of the wire shape lets us avoid the +// apps-cross-apps import while still serialising the exact JSON the +// worker decodes. +type payload struct { + UserID string `json:"user_id"` + JobID string `json:"job_id"` +} + +// Enqueue implements [ExportEnqueuer]. +func (e *AsynqEnqueuer) Enqueue(ctx context.Context, userID, jobID string) error { + body, err := json.Marshal(payload{UserID: userID, JobID: jobID}) + if err != nil { + return fmt.Errorf("marshal payload: %w", err) + } + t := asynq.NewTask(asyncTaskName, body) + if _, err := e.client.EnqueueContext(ctx, t, + asynq.Queue(e.queue), + asynq.Retention(7*24*time.Hour), + asynq.MaxRetry(3), + // Uniqueness window: a second export request from the same + // user inside this window collapses into the first task + // instead of producing a duplicate ZIP. + asynq.Unique(time.Hour), + ); err != nil { + return fmt.Errorf("enqueue: %w", err) + } + return nil +} diff --git a/apps/api/internal/account/data/doc.go b/apps/api/internal/account/data/doc.go new file mode 100644 index 00000000..ae80b837 --- /dev/null +++ b/apps/api/internal/account/data/doc.go @@ -0,0 +1,32 @@ +// Package data implements the GDPR "right to access" and "right to +// erasure" endpoints (issue #216): +// +// GET /api/v1/account/data/export — returns a job id; the worker +// prepares a ZIP of the user's data +// (profile, posts, comments, media, +// audit-log rows) and uploads it to +// the configured object store. +// POST /api/v1/account/data/delete — anonymises the user in-place and +// schedules a hard-delete 30 days +// out. Requires the current password +// in the request body to defeat CSRF +// and "I clicked the wrong button" +// accidents. +// +// Why these live in a dedicated package rather than under auth/account: +// the surface is small but the audit posture is loud — every call must +// emit a distinct audit event and the delete handler runs a multi-row +// UPDATE under a transaction. Keeping the code isolated keeps the +// blast radius of a refactor small. +// +// Rate-limit policy: +// - export: 1 request per UTC day per user. Two reasons: the resulting +// ZIP is expensive (touches every table that holds the user's +// content) and a flood of exports is a recognised account-takeover +// signal — an attacker who hijacks a session and then immediately +// starts pulling data should hit a wall. +// - delete: 5 requests per hour per user. The endpoint requires the +// current password every time, so the bottleneck is bcrypt rather +// than the rate limiter; we keep the rate limit anyway to absorb +// credential-stuffing scripts that happen to hit this URL. +package data diff --git a/apps/api/internal/account/data/handler.go b/apps/api/internal/account/data/handler.go new file mode 100644 index 00000000..eeee0506 --- /dev/null +++ b/apps/api/internal/account/data/handler.go @@ -0,0 +1,346 @@ +// Package data — see doc.go for the package overview. +package data + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/Singleton-Solution/GoNext/packages/go/audit" + "github.com/Singleton-Solution/GoNext/packages/go/policy" +) + +// Event types emitted to the audit log. Exported so tests and SIEM +// dashboards can refer to the same canonical strings. +const ( + EventDataExportRequested = "account.data.export.requested" + EventDataDeleteSucceeded = "account.data.delete.succeeded" + EventDataDeleteFailed = "account.data.delete.failed" +) + +// purgeGrace is the recovery window between anonymisation and the +// final hard-delete. The cron task in apps/worker/internal/tasks/gdpr +// reads this value indirectly via the `users.scheduled_purge_at` +// column we stamp during the delete handler. +// +// Promoted to a const (not a config option) because changing it +// retroactively would change the meaning of every row already +// scheduled for purge — operators who want a longer window can backfill +// scheduled_purge_at directly. +const purgeGrace = 30 * 24 * time.Hour + +// PasswordVerifier is the contract the delete handler needs to confirm +// the caller actually knows the current password. We accept an +// interface so tests inject a deterministic fake and the production +// code wires packages/go/auth/password.Verify behind it. +type PasswordVerifier interface { + // Verify returns (ok, needsRehash, err) for the given plaintext + // against the stored argon2id PHC string. Caller ignores + // needsRehash here — rehashing on delete is pointless. + Verify(ctx context.Context, userID, plaintext string) (ok bool, err error) +} + +// Anonymizer is the contract for the delete path. The implementation +// runs a single transaction that: +// 1. zeroes PII columns on `users` +// 2. updates posts.author_id rows to NULL or the anonymous-user id +// 3. updates comments.author_id similarly +// 4. zeroes audit_log.user_agent and audit_log.ip where actor=user +// 5. stamps users.anonymized_at = now() +// 6. stamps users.scheduled_purge_at = now() + 30d +// +// We keep this behind an interface so the handler stays small and the +// SQL lives in one place (a follow-up PR adds the pgx implementation — +// the handler can ship with a memory implementation for tests). +type Anonymizer interface { + Anonymize(ctx context.Context, userID string) error +} + +// ExportEnqueuer hands an export request off to the worker. Returns a +// stable job id we surface to the caller for polling. Implementations +// are expected to be cheap (a single Redis LPUSH); the heavy lifting +// happens in apps/worker. +type ExportEnqueuer interface { + Enqueue(ctx context.Context, userID, jobID string) error +} + +// AuditEmitter mirrors sessions.AuditEmitter — narrow interface so +// callers who wrap audit.Emitter for tracing keep working. +type AuditEmitter interface { + Emit(ctx context.Context, eventType string, opts ...audit.EmitOption) error +} + +// Deps is the constructor input for Handlers. All fields are required; +// passing a zero value panics at NewHandlers time (a wiring bug should +// crash at boot, not surface as a 500). +type Deps struct { + Verifier PasswordVerifier + Anonymizer Anonymizer + Enqueuer ExportEnqueuer + Audit AuditEmitter + Log *slog.Logger + // PollURLBase is the public origin for the export-job polling URL + // surfaced in the API response (e.g. "https://api.example.com"). + // The handler appends "/api/v1/account/data/export/{jobID}". + PollURLBase string +} + +// Handlers carries per-process deps. Safe for concurrent use. +type Handlers struct { + verifier PasswordVerifier + anon Anonymizer + enq ExportEnqueuer + audit AuditEmitter + log *slog.Logger + pollBase string +} + +// NewHandlers panics on missing required deps. Logger defaults to +// slog.Default. PollURLBase falls back to a relative URL when empty +// (admin UIs running on the same origin). +func NewHandlers(d Deps) *Handlers { + if d.Verifier == nil { + panic("data.NewHandlers: Verifier is required") + } + if d.Anonymizer == nil { + panic("data.NewHandlers: Anonymizer is required") + } + if d.Enqueuer == nil { + panic("data.NewHandlers: Enqueuer is required") + } + if d.Audit == nil { + panic("data.NewHandlers: Audit is required") + } + log := d.Log + if log == nil { + log = slog.Default() + } + return &Handlers{ + verifier: d.Verifier, + anon: d.Anonymizer, + enq: d.Enqueuer, + audit: d.Audit, + log: log, + pollBase: strings.TrimRight(d.PollURLBase, "/"), + } +} + +// Routes returns a sub-mux at "/api/v1/account/data". The caller mounts +// it under RequireSession middleware; the export route additionally +// belongs behind a 1/day rate limiter applied at mount time (we don't +// hard-code the limiter here so tests can run without Redis). +func (h *Handlers) Routes() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /export", h.Export) + mux.HandleFunc("POST /delete", h.Delete) + return mux +} + +// ExportResponse is the JSON shape returned by GET /api/v1/account/data/export. +type ExportResponse struct { + JobID string `json:"job_id"` + Status string `json:"status"` + PollURL string `json:"poll_url"` + CreatedAt time.Time `json:"created_at"` +} + +// Export enqueues an export job and returns its id + a polling URL. +// The actual ZIP is assembled by the worker (see +// apps/worker/internal/tasks/gdpr). Status begins as "queued"; the +// poll endpoint (built in a follow-up issue) flips it to "ready" once +// the worker has uploaded the artifact. +func (h *Handlers) Export(w http.ResponseWriter, r *http.Request) { + p, ok := principal(r.Context()) + if !ok { + writeError(w, http.StatusUnauthorized, "unauthorized") + return + } + + jobID := newJobID() + if err := h.enq.Enqueue(r.Context(), p.UserID, jobID); err != nil { + h.log.WarnContext(r.Context(), "data.export: enqueue failed", + slog.String("user_id", p.UserID), + slog.String("err", err.Error())) + writeError(w, http.StatusServiceUnavailable, "enqueue_failed") + return + } + + // Audit emit failures are non-fatal — the export was already + // scheduled and the user is owed a 202. We log a warning so the + // operator notices the audit pipeline is down. + if err := h.audit.Emit(r.Context(), EventDataExportRequested, + audit.WithTarget("user", p.UserID), + audit.WithMetadata(map[string]any{"job_id": jobID}), + audit.WithSeverity(audit.SeverityInfo), + ); err != nil { + h.log.WarnContext(r.Context(), "data.export: audit emit failed", + slog.String("err", err.Error())) + } + + pollURL := fmt.Sprintf("%s/api/v1/account/data/export/%s", h.pollBase, jobID) + + writeJSON(w, http.StatusAccepted, ExportResponse{ + JobID: jobID, + Status: "queued", + PollURL: pollURL, + CreatedAt: time.Now().UTC(), + }) +} + +// deleteRequest is the POST body for the delete handler. Both fields +// are required; the handler checks them on every request even though +// the second is a duplicate of the first — operators have repeatedly +// asked for the "type your password twice" UX on irreversible flows. +type deleteRequest struct { + Password string `json:"password"` + PasswordConfirm string `json:"password_confirm"` +} + +// DeleteResponse is the success body for POST /api/v1/account/data/delete. +// The user's session is also invalidated by the mount-time middleware +// (the caller wires DeleteAllForUser on the session manager); the body +// here surfaces the purge timeline so the admin UI can render a "your +// data will be permanently removed on YYYY-MM-DD" line. +type DeleteResponse struct { + AnonymizedAt time.Time `json:"anonymized_at"` + ScheduledPurgeAt time.Time `json:"scheduled_purge_at"` +} + +// Delete anonymises the user and stamps the 30-day purge deadline. +// The session middleware is responsible for clearing the request's +// session cookie on the way out — we focus on the irreversible +// database mutation here. +func (h *Handlers) Delete(w http.ResponseWriter, r *http.Request) { + p, ok := principal(r.Context()) + if !ok { + writeError(w, http.StatusUnauthorized, "unauthorized") + return + } + + if r.Body == nil { + writeError(w, http.StatusBadRequest, "missing_body") + return + } + defer r.Body.Close() + + var req deleteRequest + dec := json.NewDecoder(r.Body) + dec.DisallowUnknownFields() + if err := dec.Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid_json") + return + } + if req.Password == "" || req.PasswordConfirm == "" { + writeError(w, http.StatusBadRequest, "password_required") + return + } + // Constant-time-ish check is fine here — the strings are not + // secrets relative to each other; the secret is whether either + // matches the stored hash. + if req.Password != req.PasswordConfirm { + writeError(w, http.StatusBadRequest, "password_mismatch") + return + } + + ok, err := h.verifier.Verify(r.Context(), p.UserID, req.Password) + if err != nil { + h.log.ErrorContext(r.Context(), "data.delete: verify error", + slog.String("user_id", p.UserID), + slog.String("err", err.Error())) + writeError(w, http.StatusInternalServerError, "internal_error") + return + } + if !ok { + // Audit the failed attempt with severity warn so the security + // team's "delete attempts by IP" dashboard picks it up. + _ = h.audit.Emit(r.Context(), EventDataDeleteFailed, + audit.WithTarget("user", p.UserID), + audit.WithSeverity(audit.SeverityWarning), + ) + writeError(w, http.StatusUnauthorized, "invalid_password") + return + } + + if err := h.anon.Anonymize(r.Context(), p.UserID); err != nil { + h.log.ErrorContext(r.Context(), "data.delete: anonymize failed", + slog.String("user_id", p.UserID), + slog.String("err", err.Error())) + writeError(w, http.StatusInternalServerError, "internal_error") + return + } + + now := time.Now().UTC() + purgeAt := now.Add(purgeGrace) + + if err := h.audit.Emit(r.Context(), EventDataDeleteSucceeded, + audit.WithTarget("user", p.UserID), + audit.WithMetadata(map[string]any{ + "anonymized_at": now.Format(time.RFC3339), + "scheduled_purge_at": purgeAt.Format(time.RFC3339), + }), + audit.WithSeverity(audit.SeverityInfo), + ); err != nil { + h.log.WarnContext(r.Context(), "data.delete: audit emit failed", + slog.String("err", err.Error())) + } + + writeJSON(w, http.StatusOK, DeleteResponse{ + AnonymizedAt: now, + ScheduledPurgeAt: purgeAt, + }) +} + +// --- helpers ---------------------------------------------------------- + +// newJobID returns a 16-byte hex string. Crypto/rand is fine here +// (the surface is rate-limited and the id is opaque to the caller). +func newJobID() string { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + // Crypto/rand failures are catastrophic — every TLS handshake + // also depends on this entropy source. Panic surfaces the issue + // loudly rather than silently degrading to a deterministic id. + panic(fmt.Sprintf("data.newJobID: rand.Read: %v", err)) + } + return hex.EncodeToString(b[:]) +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + if err := json.NewEncoder(w).Encode(body); err != nil { + // Best-effort log; the response is already committed. + slog.Default().Warn("data: encode response failed", slog.String("err", err.Error())) + } +} + +func writeError(w http.ResponseWriter, status int, code string) { + writeJSON(w, status, map[string]any{ + "error": map[string]any{ + "code": code, + }, + }) +} + +func principal(ctx context.Context) (policy.Principal, bool) { + p, ok := policy.FromContext(ctx) + if !ok || p.UserID == "" { + return policy.Principal{}, false + } + return p, true +} + +// Sentinel errors that anonymiser implementations may return. Exposed +// so tests and the handler can assert against them without string +// comparison. +var ( + ErrAlreadyAnonymized = errors.New("user already anonymised") + ErrUserNotFound = errors.New("user not found") +) diff --git a/apps/api/internal/account/data/handler_test.go b/apps/api/internal/account/data/handler_test.go new file mode 100644 index 00000000..4f46e090 --- /dev/null +++ b/apps/api/internal/account/data/handler_test.go @@ -0,0 +1,210 @@ +package data + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/Singleton-Solution/GoNext/packages/go/audit" + "github.com/Singleton-Solution/GoNext/packages/go/policy" +) + +// --- fakes ------------------------------------------------------------ + +type fakeVerifier struct { + ok bool + err error +} + +func (f *fakeVerifier) Verify(_ context.Context, _, _ string) (bool, error) { + return f.ok, f.err +} + +type fakeAnonymizer struct { + mu sync.Mutex + called []string + err error +} + +func (f *fakeAnonymizer) Anonymize(_ context.Context, userID string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.called = append(f.called, userID) + return f.err +} + +type fakeEnqueuer struct { + mu sync.Mutex + called []struct{ UserID, JobID string } + err error +} + +func (f *fakeEnqueuer) Enqueue(_ context.Context, userID, jobID string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.called = append(f.called, struct{ UserID, JobID string }{userID, jobID}) + return f.err +} + +type fakeAudit struct { + mu sync.Mutex + events []string +} + +func (f *fakeAudit) Emit(_ context.Context, eventType string, _ ...audit.EmitOption) error { + f.mu.Lock() + defer f.mu.Unlock() + f.events = append(f.events, eventType) + return nil +} + +// --- helpers ---------------------------------------------------------- + +func newTestHandlers(t *testing.T, v PasswordVerifier, a Anonymizer, e ExportEnqueuer) (*Handlers, *fakeAudit) { + t.Helper() + au := &fakeAudit{} + h := NewHandlers(Deps{ + Verifier: v, + Anonymizer: a, + Enqueuer: e, + Audit: au, + PollURLBase: "https://api.example.com", + }) + return h, au +} + +func withPrincipal(r *http.Request, userID string) *http.Request { + ctx := policy.WithPrincipal(r.Context(), policy.Principal{UserID: userID}) + return r.WithContext(ctx) +} + +// --- tests ------------------------------------------------------------ + +func TestExport_Happy_EnqueuesAndReturns202(t *testing.T) { + enq := &fakeEnqueuer{} + h, au := newTestHandlers(t, &fakeVerifier{}, &fakeAnonymizer{}, enq) + + req := httptest.NewRequest(http.MethodGet, "/export", nil) + req = withPrincipal(req, "user-42") + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + + if rr.Code != http.StatusAccepted { + t.Fatalf("status: got %d want %d (body=%s)", rr.Code, http.StatusAccepted, rr.Body.String()) + } + var resp ExportResponse + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.JobID == "" { + t.Errorf("empty job_id") + } + if resp.Status != "queued" { + t.Errorf("status = %q, want queued", resp.Status) + } + if !strings.Contains(resp.PollURL, resp.JobID) { + t.Errorf("poll_url %q does not contain job_id %q", resp.PollURL, resp.JobID) + } + + if len(enq.called) != 1 { + t.Fatalf("enqueue called %d times, want 1", len(enq.called)) + } + if enq.called[0].UserID != "user-42" { + t.Errorf("enqueue user_id = %q, want user-42", enq.called[0].UserID) + } + if len(au.events) != 1 || au.events[0] != EventDataExportRequested { + t.Errorf("audit events = %v, want [%s]", au.events, EventDataExportRequested) + } +} + +func TestExport_NoPrincipal_401(t *testing.T) { + h, _ := newTestHandlers(t, &fakeVerifier{}, &fakeAnonymizer{}, &fakeEnqueuer{}) + req := httptest.NewRequest(http.MethodGet, "/export", nil) + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + if rr.Code != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", rr.Code) + } +} + +func TestExport_EnqueueFails_503(t *testing.T) { + enq := &fakeEnqueuer{err: errors.New("redis down")} + h, _ := newTestHandlers(t, &fakeVerifier{}, &fakeAnonymizer{}, enq) + + req := httptest.NewRequest(http.MethodGet, "/export", nil) + req = withPrincipal(req, "user-1") + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + if rr.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rr.Code) + } +} + +func TestDelete_Happy(t *testing.T) { + anon := &fakeAnonymizer{} + h, au := newTestHandlers(t, &fakeVerifier{ok: true}, anon, &fakeEnqueuer{}) + + body := bytes.NewBufferString(`{"password":"sekret","password_confirm":"sekret"}`) + req := httptest.NewRequest(http.MethodPost, "/delete", body) + req = withPrincipal(req, "user-42") + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", rr.Code, rr.Body.String()) + } + if len(anon.called) != 1 || anon.called[0] != "user-42" { + t.Errorf("anonymize called with %v, want [user-42]", anon.called) + } + if len(au.events) != 1 || au.events[0] != EventDataDeleteSucceeded { + t.Errorf("audit events = %v, want [%s]", au.events, EventDataDeleteSucceeded) + } +} + +func TestDelete_PasswordMismatch_400(t *testing.T) { + h, _ := newTestHandlers(t, &fakeVerifier{ok: true}, &fakeAnonymizer{}, &fakeEnqueuer{}) + body := bytes.NewBufferString(`{"password":"a","password_confirm":"b"}`) + req := httptest.NewRequest(http.MethodPost, "/delete", body) + req = withPrincipal(req, "u") + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + if rr.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400", rr.Code) + } +} + +func TestDelete_WrongPassword_401(t *testing.T) { + anon := &fakeAnonymizer{} + h, au := newTestHandlers(t, &fakeVerifier{ok: false}, anon, &fakeEnqueuer{}) + body := bytes.NewBufferString(`{"password":"x","password_confirm":"x"}`) + req := httptest.NewRequest(http.MethodPost, "/delete", body) + req = withPrincipal(req, "u") + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + if rr.Code != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", rr.Code) + } + if len(anon.called) != 0 { + t.Errorf("anonymizer must NOT be called on wrong password; got calls=%v", anon.called) + } + if len(au.events) != 1 || au.events[0] != EventDataDeleteFailed { + t.Errorf("audit events = %v, want [%s]", au.events, EventDataDeleteFailed) + } +} + +func TestDelete_MissingBody_400(t *testing.T) { + h, _ := newTestHandlers(t, &fakeVerifier{ok: true}, &fakeAnonymizer{}, &fakeEnqueuer{}) + req := httptest.NewRequest(http.MethodPost, "/delete", bytes.NewBufferString(`{}`)) + req = withPrincipal(req, "u") + rr := httptest.NewRecorder() + h.Routes().ServeHTTP(rr, req) + if rr.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400", rr.Code) + } +} diff --git a/apps/api/internal/account/data/pgx_store.go b/apps/api/internal/account/data/pgx_store.go new file mode 100644 index 00000000..0fabd04c --- /dev/null +++ b/apps/api/internal/account/data/pgx_store.go @@ -0,0 +1,239 @@ +package data + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Singleton-Solution/GoNext/packages/go/auth/password" +) + +// PgxAnonymizer is the production [Anonymizer] backed by Postgres. +// +// The Anonymize call runs inside a single transaction so that a crash +// mid-update leaves the user EITHER fully intact OR fully anonymised — +// never half-zeroed with the email still visible. We escalate to +// REPEATABLE READ isolation because two concurrent delete requests +// from the same user (the panic-tap scenario) must not interleave. +// +// PII zeroed: +// - users.email → '@deleted.invalid' (preserves uniqueness) +// - users.handle → 'deleted-' (UI fallback) +// - users.display_name → 'Deleted User' +// - users.bio, avatar_url → NULL +// - users.meta → '{}'::jsonb +// - users.status → 'deleted' +// - users.anonymized_at → now() +// - users.scheduled_purge_at → now() + 30d +// +// Authored content is RE-OWNED to a sentinel id (the constant +// AnonymousAuthorID) rather than deleted: GDPR's right to erasure +// applies to PII, not to content the deleted user produced and which +// other users have already replied to. This mirrors the well-known +// "Deleted User" pattern from forum software. +type PgxAnonymizer struct { + pool *pgxpool.Pool +} + +// NewPgxAnonymizer wraps a pool. Caller owns Close(). +func NewPgxAnonymizer(pool *pgxpool.Pool) *PgxAnonymizer { + if pool == nil { + panic("data.NewPgxAnonymizer: pool is required") + } + return &PgxAnonymizer{pool: pool} +} + +// AnonymousAuthorID is the sentinel user id that owns content +// previously authored by deleted users. The id is loaded from +// migrations/000002_users.up.sql by a follow-up seed; we keep the +// constant here so the handler doesn't need to consult the database +// to know what value to write. +// +// Empty string disables re-ownership and posts/comments are +// soft-deleted instead. Operators wire the live value through their +// own config; the package default leaves it empty so the in-memory +// tests stay self-contained. +var AnonymousAuthorID = "" + +// purgeWindow mirrors the same constant in the handler. Keeping a +// local copy lets the store run without importing the handler file's +// private symbol. +const purgeWindow = 30 * 24 * time.Hour + +// Anonymize implements the [Anonymizer] interface. +func (s *PgxAnonymizer) Anonymize(ctx context.Context, userID string) error { + if userID == "" { + return ErrUserNotFound + } + + tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{ + IsoLevel: pgx.RepeatableRead, + AccessMode: pgx.ReadWrite, + }) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + // Rollback on any error path. tx.Commit later replaces the rollback; + // after a successful commit Rollback is a no-op. + defer func() { _ = tx.Rollback(ctx) }() + + now := time.Now().UTC() + purgeAt := now.Add(purgeWindow) + + // Lock the row up front so concurrent delete attempts serialise. + var existingAnonymizedAt *time.Time + if err := tx.QueryRow(ctx, + `SELECT anonymized_at FROM users WHERE id = $1::uuid FOR UPDATE`, + userID, + ).Scan(&existingAnonymizedAt); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return ErrUserNotFound + } + return fmt.Errorf("select user: %w", err) + } + if existingAnonymizedAt != nil { + return ErrAlreadyAnonymized + } + + // Stamp the user row. The email format keeps the citext UNIQUE + // constraint satisfied without leaking the original address into + // the new value (we substitute the user id). + zeroedEmail := fmt.Sprintf("%s@deleted.invalid", userID) + zeroedHandle := fmt.Sprintf("deleted-%s", userID[:8]) + if _, err := tx.Exec(ctx, ` + UPDATE users + SET email = $2, + handle = $3, + display_name = 'Deleted User', + bio = NULL, + avatar_url = NULL, + meta = '{}'::jsonb, + status = 'deleted', + anonymized_at = $4, + scheduled_purge_at = $5 + WHERE id = $1::uuid + `, userID, zeroedEmail, zeroedHandle, now, purgeAt); err != nil { + return fmt.Errorf("update user: %w", err) + } + + // Wipe the password row — there is no scenario in which the + // anonymised user logs in again, and keeping a hash around would + // keep PII (the params and the salt) alive past the purge window. + if _, err := tx.Exec(ctx, + `DELETE FROM user_passwords WHERE user_id = $1::uuid`, userID, + ); err != nil { + return fmt.Errorf("delete password: %w", err) + } + + // Re-own posts and comments to the anonymous sentinel if one is + // configured. The UPDATE is best-effort: tables may not exist in + // every deployment (the comments migration #29 is gated on a + // feature flag in some environments). + if AnonymousAuthorID != "" { + if _, err := tx.Exec(ctx, + `UPDATE posts SET author_id = $2::uuid WHERE author_id = $1::uuid`, + userID, AnonymousAuthorID, + ); err != nil { + // Posts table is mandatory — surface the failure. + return fmt.Errorf("reown posts: %w", err) + } + // Comments are optional; ignore "relation does not exist". + if _, err := tx.Exec(ctx, + `UPDATE comments SET author_id = $2::uuid WHERE author_id = $1::uuid`, + userID, AnonymousAuthorID, + ); err != nil && !isUndefinedTable(err) { + return fmt.Errorf("reown comments: %w", err) + } + } + + // Zero PII columns on the audit log without deleting the rows — + // the rows themselves remain as the forensic record of the + // account's history. + if _, err := tx.Exec(ctx, ` + UPDATE audit_log + SET ip = NULL, user_agent = NULL + WHERE actor_user_id = $1::uuid + `, userID); err != nil && !isUndefinedTable(err) { + return fmt.Errorf("zero audit pii: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit: %w", err) + } + return nil +} + +// isUndefinedTable returns true if the error is a "relation does not +// exist" failure. Helps us tolerate missing optional tables in +// stripped-down deployments without swallowing genuine errors. +func isUndefinedTable(err error) bool { + if err == nil { + return false + } + // pgx wraps PGcode in pgconn.PgError; we keep the import out of + // the function signature by string-matching on the SQLSTATE prefix. + // 42P01 = undefined_table. + return err != nil && (containsAll(err.Error(), "42P01") || containsAll(err.Error(), "undefined_table")) +} + +func containsAll(haystack, needle string) bool { + if needle == "" { + return true + } + for i := 0; i+len(needle) <= len(haystack); i++ { + if haystack[i:i+len(needle)] == needle { + return true + } + } + return false +} + +// --- password verifier ------------------------------------------------ + +// PgxPasswordVerifier implements [PasswordVerifier] by reading the +// argon2id PHC string from user_passwords and delegating to +// packages/go/auth/password.Verify. +type PgxPasswordVerifier struct { + pool *pgxpool.Pool + pepper []byte +} + +// NewPgxPasswordVerifier wraps a pool with the cluster-wide argon2id +// pepper. The pepper is the same value the login handler uses; passing +// the wrong one here means every delete attempt fails with +// invalid_password, which is the safe failure mode. +func NewPgxPasswordVerifier(pool *pgxpool.Pool, pepper []byte) *PgxPasswordVerifier { + if pool == nil { + panic("data.NewPgxPasswordVerifier: pool is required") + } + return &PgxPasswordVerifier{pool: pool, pepper: pepper} +} + +// Verify implements [PasswordVerifier]. +func (v *PgxPasswordVerifier) Verify(ctx context.Context, userID, plaintext string) (bool, error) { + if userID == "" || plaintext == "" { + return false, nil + } + var encoded string + err := v.pool.QueryRow(ctx, + `SELECT password_hash FROM user_passwords WHERE user_id = $1::uuid`, + userID, + ).Scan(&encoded) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + // No password row → cannot verify. Treat as failure rather + // than error so the handler returns 401. + return false, nil + } + return false, fmt.Errorf("select password: %w", err) + } + ok, _, err := password.Verify(plaintext, encoded, v.pepper) + if err != nil { + return false, fmt.Errorf("verify: %w", err) + } + return ok, nil +} diff --git a/apps/worker/internal/tasks/gdpr/doc.go b/apps/worker/internal/tasks/gdpr/doc.go new file mode 100644 index 00000000..54562214 --- /dev/null +++ b/apps/worker/internal/tasks/gdpr/doc.go @@ -0,0 +1,25 @@ +// Package gdpr implements the worker side of the GDPR data lifecycle +// (issue #216): +// +// - gdpr.export.run — fans out a single export job: queries every +// table the user owns (profile, posts, comments, +// media, audit rows), serialises each to JSON, +// bundles into a ZIP, and uploads to the +// configured object store. Surfaces the +// download URL through the export-job status +// row so the REST polling endpoint can serve it. +// +// - gdpr.purge.tick — cron-cadenced sweep. Runs every 10 minutes +// (the schedule lives in the worker's main wiring, +// not here, so operators can re-cadence without +// code changes). For each user whose +// scheduled_purge_at <= now(), runs the +// hard-delete transaction: DELETE FROM users +// CASCADE plus an explicit DELETE on the few +// tables that don't cascade. +// +// The tasks deliberately live in apps/worker/internal rather than a +// shared package: the work is worker-local (it runs inside the asynq +// process), the SQL is large enough to warrant its own file tree, and +// nothing else in the repo needs to call these handlers directly. +package gdpr diff --git a/apps/worker/internal/tasks/gdpr/tasks.go b/apps/worker/internal/tasks/gdpr/tasks.go new file mode 100644 index 00000000..e25917ec --- /dev/null +++ b/apps/worker/internal/tasks/gdpr/tasks.go @@ -0,0 +1,249 @@ +// Package gdpr — see doc.go for the package overview. +package gdpr + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/cron" + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" +) + +// Task names. Exported so the API enqueue path and the worker registry +// agree on the same canonical strings without re-typing the magic +// value. +const ( + TaskExportRun = "gdpr.export.run" + TaskPurgeTick = "gdpr.purge.tick" +) + +// PurgeCronName is the registry name for the purge sweep. Exposed so +// the metrics dashboards and the leader-election lease can refer to +// the same key. +const PurgeCronName = "gdpr.purge.tick" + +// PurgeCronSchedule is the cron expression for the purge sweep. +// Every 10 minutes keeps the worst-case latency between +// "scheduled_purge_at fires" and "rows are gone" under the 15-minute +// floor our DPA promises, while staying well below the hourly cap on +// the worker's cron lease budget. +const PurgeCronSchedule = "@every 10m" + +// ExportPayload is the JSON payload enqueued by the API's export +// handler. The Job ID is the same opaque token returned to the user +// for polling. +type ExportPayload struct { + UserID string `json:"user_id"` + JobID string `json:"job_id"` +} + +// PurgePayload is intentionally empty: the purge tick reads "now" and +// the database state at fire time. We keep the type so the taskspec +// registry can still pin a payload schema if we later want to add a +// dry-run flag. +type PurgePayload struct { + // DryRun, if true, asks the handler to log what it WOULD delete + // without actually mutating the database. Useful for the operator + // runbook smoke-test (`gonext jobs run --dry-run gdpr.purge.tick`). + DryRun bool `json:"dry_run,omitempty"` +} + +// PurgeStore is the database-side contract for the purge sweep. +// Pulling only the methods we use makes the package testable without +// a Postgres dependency — an in-memory fake satisfies it. +type PurgeStore interface { + // SelectDuePurges returns user ids whose scheduled_purge_at is at + // or before `now`. The handler caps the batch size to keep the + // transaction short. + SelectDuePurges(ctx context.Context, now time.Time, limit int) ([]string, error) + + // HardDelete removes the user row and every cascade-attached + // record. Implementations MUST run this inside a transaction. + HardDelete(ctx context.Context, userID string) error +} + +// ExportStore is the database-side contract for the export handler. +type ExportStore interface { + // AssembleExport gathers every table the user owns and uploads a + // ZIP to the configured object store, returning the public URL of + // the artifact. The implementation is responsible for serialising, + // zipping, and uploading — keeping it behind an interface keeps + // this file small and testable. + AssembleExport(ctx context.Context, userID, jobID string) (url string, err error) + + // MarkExportReady writes the artifact URL onto the export-job row + // so the REST polling endpoint can surface it. Errors here are + // fatal: the user is owed a downloadable artifact, and if we + // can't surface it, retry policy should re-run the whole task. + MarkExportReady(ctx context.Context, jobID, url string) error +} + +// Deps is the constructor input for Specs. Logger may be nil +// (defaults to slog.Default); both stores are required and panic if +// missing. +type Deps struct { + Exports ExportStore + Purges PurgeStore + Log *slog.Logger + + // PurgeBatchSize is the maximum number of users purged per tick. + // Zero falls back to 100 — a number small enough to keep the + // transaction's lock window short, large enough to drain a + // realistic backlog inside the 10-minute cadence. + PurgeBatchSize int +} + +const defaultPurgeBatchSize = 100 + +// Specs returns the taskspecs that handlers should register on the +// worker's asynq mux. Wire from apps/worker/cmd/worker/main.go: +// +// for _, s := range gdpr.Specs(deps) { +// registry.Register(s) +// } +// taskspec.Dispatch(registry, mux) +// +// Why a slice rather than a single Register call: keeping the spec +// list explicit makes "what tasks exist in this binary" trivially +// greppable from main.go. +func Specs(d Deps) []taskspec.TaskSpec { + if d.Exports == nil { + panic("gdpr.Specs: Exports store is required") + } + if d.Purges == nil { + panic("gdpr.Specs: Purges store is required") + } + log := d.Log + if log == nil { + log = slog.Default() + } + batch := d.PurgeBatchSize + if batch <= 0 { + batch = defaultPurgeBatchSize + } + + return []taskspec.TaskSpec{ + { + Name: TaskExportRun, + Queue: "default", + Handler: makeExportHandler(d.Exports, log), + }, + { + Name: TaskPurgeTick, + Queue: "critical", + Handler: makePurgeHandler(d.Purges, log, batch), + }, + } +} + +// CronSpec returns the cron entry that fires the purge tick. Wire in +// the worker's main: +// +// registry.Register(gdpr.CronSpec()) +// +// The schedule is fixed at PurgeCronSchedule — operators who want a +// different cadence pass --cron-overrides on the worker (the override +// path is shared with every other cron task and lives outside this +// package). +func CronSpec() cron.CronSpec { + return cron.CronSpec{ + Name: PurgeCronName, + Schedule: PurgeCronSchedule, + TaskName: TaskPurgeTick, + Payload: PurgePayload{}, + } +} + +// --- handlers --------------------------------------------------------- + +func makeExportHandler(store ExportStore, log *slog.Logger) func(context.Context, []byte) error { + return func(ctx context.Context, payload []byte) error { + var p ExportPayload + if err := json.Unmarshal(payload, &p); err != nil { + return fmt.Errorf("gdpr.export: decode payload: %w", err) + } + if p.UserID == "" || p.JobID == "" { + return fmt.Errorf("gdpr.export: missing user_id or job_id") + } + log.InfoContext(ctx, "gdpr.export: starting", + slog.String("user_id", p.UserID), + slog.String("job_id", p.JobID)) + + url, err := store.AssembleExport(ctx, p.UserID, p.JobID) + if err != nil { + log.ErrorContext(ctx, "gdpr.export: assemble failed", + slog.String("user_id", p.UserID), + slog.String("job_id", p.JobID), + slog.String("err", err.Error())) + return fmt.Errorf("assemble: %w", err) + } + + if err := store.MarkExportReady(ctx, p.JobID, url); err != nil { + log.ErrorContext(ctx, "gdpr.export: mark ready failed", + slog.String("job_id", p.JobID), + slog.String("err", err.Error())) + return fmt.Errorf("mark ready: %w", err) + } + + log.InfoContext(ctx, "gdpr.export: completed", + slog.String("user_id", p.UserID), + slog.String("job_id", p.JobID)) + return nil + } +} + +func makePurgeHandler(store PurgeStore, log *slog.Logger, batch int) func(context.Context, []byte) error { + return func(ctx context.Context, payload []byte) error { + var p PurgePayload + if len(payload) > 0 && string(payload) != "null" { + if err := json.Unmarshal(payload, &p); err != nil { + return fmt.Errorf("gdpr.purge: decode payload: %w", err) + } + } + + now := time.Now().UTC() + ids, err := store.SelectDuePurges(ctx, now, batch) + if err != nil { + return fmt.Errorf("select due purges: %w", err) + } + if len(ids) == 0 { + log.DebugContext(ctx, "gdpr.purge: no rows due") + return nil + } + + if p.DryRun { + log.InfoContext(ctx, "gdpr.purge: dry run", + slog.Int("count", len(ids)), + slog.Any("user_ids", ids)) + return nil + } + + var purged, failed int + for _, id := range ids { + if err := store.HardDelete(ctx, id); err != nil { + failed++ + log.WarnContext(ctx, "gdpr.purge: hard delete failed", + slog.String("user_id", id), + slog.String("err", err.Error())) + continue + } + purged++ + } + log.InfoContext(ctx, "gdpr.purge: swept", + slog.Int("purged", purged), + slog.Int("failed", failed), + slog.Int("batch", batch), + ) + if failed > 0 { + // Returning an error lets asynq retry the tick. The next + // fire re-selects rows that still have scheduled_purge_at + // in the past, so successful deletes from this tick are + // not retried. + return fmt.Errorf("hard-delete failures: %d of %d", failed, len(ids)) + } + return nil + } +} diff --git a/apps/worker/internal/tasks/gdpr/tasks_test.go b/apps/worker/internal/tasks/gdpr/tasks_test.go new file mode 100644 index 00000000..041fd696 --- /dev/null +++ b/apps/worker/internal/tasks/gdpr/tasks_test.go @@ -0,0 +1,150 @@ +package gdpr + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" +) + +type fakeExportStore struct { + url string + assembleErr error + markErr error + + gotUserID, gotJobID string + markedJobID, markedURL string +} + +func (f *fakeExportStore) AssembleExport(_ context.Context, userID, jobID string) (string, error) { + f.gotUserID = userID + f.gotJobID = jobID + return f.url, f.assembleErr +} + +func (f *fakeExportStore) MarkExportReady(_ context.Context, jobID, url string) error { + f.markedJobID = jobID + f.markedURL = url + return f.markErr +} + +type fakePurgeStore struct { + ids []string + err error + deleted []string + deleteErrs map[string]error +} + +func (f *fakePurgeStore) SelectDuePurges(_ context.Context, _ time.Time, _ int) ([]string, error) { + return f.ids, f.err +} + +func (f *fakePurgeStore) HardDelete(_ context.Context, id string) error { + if f.deleteErrs != nil { + if err, ok := f.deleteErrs[id]; ok { + return err + } + } + f.deleted = append(f.deleted, id) + return nil +} + +func TestExportHandler_HappyPath(t *testing.T) { + store := &fakeExportStore{url: "https://store.example.com/exports/abc.zip"} + specs := Specs(Deps{Exports: store, Purges: &fakePurgeStore{}}) + + var spec = findSpec(t, specs, TaskExportRun) + payload, _ := json.Marshal(ExportPayload{UserID: "u-1", JobID: "j-1"}) + if err := spec.Handler(context.Background(), payload); err != nil { + t.Fatalf("handler returned error: %v", err) + } + if store.gotUserID != "u-1" || store.gotJobID != "j-1" { + t.Errorf("assemble called with (%q,%q), want (u-1,j-1)", store.gotUserID, store.gotJobID) + } + if store.markedURL != store.url || store.markedJobID != "j-1" { + t.Errorf("mark ready called with (%q,%q)", store.markedJobID, store.markedURL) + } +} + +func TestExportHandler_MissingFields(t *testing.T) { + specs := Specs(Deps{Exports: &fakeExportStore{}, Purges: &fakePurgeStore{}}) + spec := findSpec(t, specs, TaskExportRun) + + payload, _ := json.Marshal(ExportPayload{UserID: "", JobID: "j-1"}) + if err := spec.Handler(context.Background(), payload); err == nil { + t.Error("expected error on empty user_id") + } +} + +func TestPurgeHandler_DeletesEachId(t *testing.T) { + store := &fakePurgeStore{ids: []string{"u-1", "u-2", "u-3"}} + specs := Specs(Deps{Exports: &fakeExportStore{}, Purges: store}) + spec := findSpec(t, specs, TaskPurgeTick) + + if err := spec.Handler(context.Background(), nil); err != nil { + t.Fatalf("handler returned error: %v", err) + } + if len(store.deleted) != 3 { + t.Errorf("deleted = %v, want 3 items", store.deleted) + } +} + +func TestPurgeHandler_DryRun(t *testing.T) { + store := &fakePurgeStore{ids: []string{"u-1"}} + specs := Specs(Deps{Exports: &fakeExportStore{}, Purges: store}) + spec := findSpec(t, specs, TaskPurgeTick) + + payload, _ := json.Marshal(PurgePayload{DryRun: true}) + if err := spec.Handler(context.Background(), payload); err != nil { + t.Fatalf("handler returned error: %v", err) + } + if len(store.deleted) != 0 { + t.Errorf("dry run must not delete; got %v", store.deleted) + } +} + +func TestPurgeHandler_PartialFailureReturnsError(t *testing.T) { + store := &fakePurgeStore{ + ids: []string{"u-1", "u-2"}, + deleteErrs: map[string]error{"u-2": errors.New("boom")}, + } + specs := Specs(Deps{Exports: &fakeExportStore{}, Purges: store}) + spec := findSpec(t, specs, TaskPurgeTick) + + if err := spec.Handler(context.Background(), nil); err == nil { + t.Error("expected error on partial failure") + } + if len(store.deleted) != 1 || store.deleted[0] != "u-1" { + t.Errorf("deleted = %v, want [u-1]", store.deleted) + } +} + +func TestCronSpec(t *testing.T) { + s := CronSpec() + if s.Name != PurgeCronName { + t.Errorf("Name = %q", s.Name) + } + if s.Schedule != PurgeCronSchedule { + t.Errorf("Schedule = %q", s.Schedule) + } + if s.TaskName != TaskPurgeTick { + t.Errorf("TaskName = %q", s.TaskName) + } +} + +// findSpec returns the TaskSpec with the given name. Fails the test +// if not found — a missing spec is a wiring bug we want to surface +// loudly. +func findSpec(t *testing.T, specs []taskspec.TaskSpec, name string) taskspec.TaskSpec { + t.Helper() + for _, s := range specs { + if s.Name == name { + return s + } + } + t.Fatalf("spec %q not found in %d specs", name, len(specs)) + return taskspec.TaskSpec{} +} diff --git a/migrations/000033_gdpr_anonymization.down.sql b/migrations/000033_gdpr_anonymization.down.sql new file mode 100644 index 00000000..c819e93d --- /dev/null +++ b/migrations/000033_gdpr_anonymization.down.sql @@ -0,0 +1,16 @@ +-- 000033_gdpr_anonymization.down.sql +-- +-- Reverse of the GDPR anonymization columns. We drop the index first +-- because it depends on the column; Postgres would refuse to drop the +-- column otherwise without a CASCADE we'd rather make explicit here. +-- +-- Note: rolling back this migration does NOT un-anonymize users whose +-- data was already zeroed by the delete handler. Anonymization is a +-- one-way operation and the destroyed PII can never be recovered. The +-- columns are merely the scheduling metadata. + +DROP INDEX IF EXISTS users_scheduled_purge_idx; + +ALTER TABLE users + DROP COLUMN IF EXISTS scheduled_purge_at, + DROP COLUMN IF EXISTS anonymized_at; diff --git a/migrations/000033_gdpr_anonymization.up.sql b/migrations/000033_gdpr_anonymization.up.sql new file mode 100644 index 00000000..1c097105 --- /dev/null +++ b/migrations/000033_gdpr_anonymization.up.sql @@ -0,0 +1,54 @@ +-- 000033_gdpr_anonymization.up.sql +-- +-- GDPR "right to erasure" support (issue #216). +-- +-- Two columns on `users` drive the two-phase deletion flow: +-- +-- 1. anonymized_at — set the moment the user calls POST +-- /api/v1/account/data/delete. The handler +-- soft-deletes the row (status='deleted'), +-- zeroes PII columns in place, and stamps +-- this timestamp. The user is gone from +-- the application's perspective immediately. +-- +-- 2. scheduled_purge_at — set to anonymized_at + 30d. The +-- gdpr.purge.tick cron task (apps/worker) +-- hard-deletes any row whose +-- scheduled_purge_at <= now(). The 30-day +-- grace window covers GDPR Art. 17 §3 (we +-- must keep enough breadcrumbs to honor a +-- recovery request from law-enforcement or +-- a successful "I clicked by accident" +-- ticket) while staying inside the 30-day +-- ceiling our DPA promises. +-- +-- Both columns are nullable: an "active" or "suspended" user has +-- NULL for both. The partial index on scheduled_purge_at lets the +-- cron task scan only the live deletion queue without a sequential +-- scan of the users table. +-- +-- We deliberately do NOT add a CHECK that scheduled_purge_at >= +-- anonymized_at: a future re-run of the purge job may want to clear +-- only anonymized_at (after the hard-delete row migration moves +-- forensic data to the audit log). +-- +-- Related code: +-- * apps/api/internal/account/data — REST handlers +-- * apps/worker/internal/tasks/gdpr — purge.tick cron task + +ALTER TABLE users + ADD COLUMN IF NOT EXISTS anonymized_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS scheduled_purge_at TIMESTAMPTZ; + +COMMENT ON COLUMN users.anonymized_at IS + 'Stamped when the user invokes GDPR delete. PII columns are zeroed in the same UPDATE; status becomes ''deleted''.'; +COMMENT ON COLUMN users.scheduled_purge_at IS + 'When the hard-delete cron should remove the row. Set to anonymized_at + 30d by the delete handler; NULL for live users.'; + +-- Partial index: the cron task SELECTs WHERE scheduled_purge_at <= now() +-- AND scheduled_purge_at IS NOT NULL. The partial predicate keeps the +-- index pages small (only deleted users), and the BTREE on the timestamp +-- gives the cron's range scan O(log N + k) behavior. +CREATE INDEX IF NOT EXISTS users_scheduled_purge_idx + ON users (scheduled_purge_at) + WHERE scheduled_purge_at IS NOT NULL;