diff --git a/docs/audit-storage-migration-compatibility.md b/docs/audit-storage-migration-compatibility.md new file mode 100644 index 00000000..055976d0 --- /dev/null +++ b/docs/audit-storage-migration-compatibility.md @@ -0,0 +1,76 @@ +# Audit storage & migration compatibility (Issue #1679) + +Area: authorization / resilience — privileged workflows, audit views, errors, and +degraded-mode behavior for the `meridian-api` administrative surface. + +This change guarantees that the storage backing every privileged/operational +workflow (the RBAC guard, admin controllers, and the audit review views) is +**deterministically compatible across schema upgrades, rollbacks, repeats, and +storage failures**. + +## Design & invariants + +The contract lives in `src/audit/audit-storage.compatibility.ts` and is enforced +by the guard (`src/auth/guard/rbac/rbac.guard.ts`), the audit writer +(`src/audit/audit.service.ts`), and a new migration +(`src/database/migrations/1787400000000-audit-storage-compat.ts`). + +| # | Invariant | Enforcement | +|---|-----------|-------------| +| 1 | **Forward compatibility** — new writers never emit rows unreadable by an older reader | `assertWriteBackwardCompatible()` rejects writes that exceed the column length bounds older readers assume; every new row is stamped `schemaVersion = CURRENT_SCHEMA_VERSION` (4). | +| 2 | **Backward compatibility** — rows written by older versions stay readable | `normalizeLegacyAuditRow()` fills safe defaults for columns introduced after a row was written (`correlationId`, `chainHash`, `contributionXp`, `epochNumber`, `schemaVersion`, …). Unknown future enum `action` values degrade to `UNKNOWN_ACTION` instead of crashing the reader. | +| 3 | **Resumable & observable migrations** | The migration uses only `IF NOT EXISTS` / `IF EXISTS` DDL and records a single-row `audit_storage_checkpoint` (phase + timestamp), surfacing progress via `RAISE NOTICE`. A partial run resumes cleanly on rerun. | +| 4 | **No partial/unauthorized state** | Repeated or replayed authorization decisions share an identity key and are de-duplicated within a 5s window, so a retry cannot leave duplicate audit rows. A failed audit write degrades to a structured `audit.degraded_mode` marker while the authorization decision remains authoritative. | + +### Schema version timeline (`audit_logs`) +- `1` — base table (issue #632 audit foundation) +- `2` — `+ correlationId` (migration `1787200000000`) +- `3` — `+ AUTHORIZATION_GRANTED` / `AUTHORIZATION_DENIED` enum values (`1787300000000`) +- `4` — `+ schemaVersion` column + `audit_storage_checkpoint` table (this work) + +Legacy rows (no `schemaVersion` column) are normalized to `null` and treated as +readable by `isSchemaCompatible(null) === true`. + +## Failure behavior & compatibility impact + +- **Audit store unavailable / slow:** the authorization decision is unaffected. + A `ForbiddenException`/`UnauthorizedException` is still thrown/allowed, the + audit write is attempted once, and on failure a `audit.degraded_mode` + + `audit.write_failed` structured log pair is emitted for operator diagnosis. +- **Oversized audit field:** the write is skipped with an `audit.write_compat_skipped` + warning rather than producing a row that older readers cannot parse. The + decision still proceeds. +- **Replay / duplicate request:** collapsed to a single audit record (idempotency). +- **Rejected/stale request:** no audit row is emitted before authentication + succeeds, so a rejected token leaves no audit or partial state. + +## Migration / rollback considerations + +- Deploy the migration (`npm run migration:run`) **before** deploying this code + so the `schemaVersion` column exists when writers start stamping it. Because + all DDL is idempotent, running the migration twice (or after a partial + failure) is safe. +- Rollback: `npm run migration:revert` removes the `schemaVersion` column and + the `audit_storage_checkpoint` table. It does **not** delete any pre-existing + audit records. Older code that ignores the column continues to work. +- The new `enum` values added in `1787300000000` are intentionally not removed + on down (PostgreSQL cannot drop enum values without a column rebuild); they + are harmless if left in place. + +## Operational limitations + +- The audit write remains **best-effort**: a sustained audit-store outage means + privileged actions still succeed but are not recorded until the store + recovers. The `audit.degraded_mode` marker is the operator signal for this gap. +- In-process de-duplication is per-instance and not shared across horizontally + scaled replicas; it bounds duplicate noise within a single instance's 5s window, + not globally. Global de-duplication would require a shared store (out of scope). + +## Security assumptions + +- Audit records are non-authoritative for access control; the RBAC decision is + computed from the JWT claims and never depends on audit storage being healthy. +- `correlationId` is treated as operator-provided untrusted input for length + bounding only; it is never used to relax authorization. +- Normalization never elevates privilege: unknown `action` values are surfaced + for review, never mapped to a granted/denied authorization outcome. diff --git a/meridian-api/jest.setup.ts b/meridian-api/jest.setup.ts index 93c9fbeb..ba4727e5 100644 --- a/meridian-api/jest.setup.ts +++ b/meridian-api/jest.setup.ts @@ -64,6 +64,35 @@ jest.mock( { virtual: true }, ); +// ----- Auth metadata decorators (aliased path style) ----- +// Controllers import these via `src/...`; replicate their real SetMetadata +// behavior so Reflector-based metadata assertions (protected-endpoints.spec) +// resolve and work without a global `src/` moduleNameMapper. +const rbacSetMetadata = jest.requireActual('@nestjs/common').SetMetadata; +jest.mock( + 'src/auth/decorators/roles/roles.decorator', + () => ({ + RequireRoles: (...roles: unknown[]) => + rbacSetMetadata('requiredRoles', roles), + }), + { virtual: true }, +); +jest.mock( + 'src/auth/decorators/permissions/permissions.decorator', + () => ({ + RequirePermissions: (...permissions: unknown[]) => + rbacSetMetadata('requiredPermissions', permissions), + }), + { virtual: true }, +); +jest.mock( + 'src/auth/decorators/public/public.decorator', + () => ({ + Public: () => rbacSetMetadata('isPublic', true), + }), + { virtual: true }, +); + // ----- Auth providers (idempotent stubs; per-spec files override as needed) ----- jest.mock( 'src/auth/providers/hashing', @@ -104,51 +133,33 @@ jest.mock( ); // ----- Entities (aliased paths) ----- -jest.mock( - 'src/users/user.entity', - () => ({ User: class User {} }), - { virtual: true }, -); -jest.mock( - 'src/post/post.entity', - () => ({ Post: class Post {} }), - { virtual: true }, -); -jest.mock( - 'src/tweets/dto/tweet.entity', - () => ({ Tweet: class Tweet {} }), - { virtual: true }, -); +jest.mock('src/users/user.entity', () => ({ User: class User {} }), { + virtual: true, +}); +jest.mock('src/post/post.entity', () => ({ Post: class Post {} }), { + virtual: true, +}); +jest.mock('src/tweets/dto/tweet.entity', () => ({ Tweet: class Tweet {} }), { + virtual: true, +}); jest.mock( 'src/tweets/entities/tweet.entity', () => ({ Tweet: class Tweet {} }), { virtual: true }, ); -jest.mock( - 'src/tag/tag.entity', - () => ({ Tag: class Tag {} }), - { virtual: true }, -); -jest.mock( - 'src/metaoption/metaoption.entity', - () => ({}), - { virtual: true }, -); -jest.mock( - 'src/metaoption/dto/create-post-meta-options.dto', - () => ({}), - { virtual: true }, -); -jest.mock( - 'src/metaoption/dto/update-post-meta-options.dto', - () => ({}), - { virtual: true }, -); -jest.mock( - 'src/metaoption/metaoption.controller', - () => ({}), - { virtual: true }, -); +jest.mock('src/tag/tag.entity', () => ({ Tag: class Tag {} }), { + virtual: true, +}); +jest.mock('src/metaoption/metaoption.entity', () => ({}), { virtual: true }); +jest.mock('src/metaoption/dto/create-post-meta-options.dto', () => ({}), { + virtual: true, +}); +jest.mock('src/metaoption/dto/update-post-meta-options.dto', () => ({}), { + virtual: true, +}); +jest.mock('src/metaoption/metaoption.controller', () => ({}), { + virtual: true, +}); // ----- Services referenced through aliased paths ----- jest.mock( @@ -202,19 +213,15 @@ jest.mock( () => ({ PatchPostDto: class PatchPostDto {} }), { virtual: true }, ); -jest.mock( - 'src/DTO/getPostdto', - () => ({ GetPostsDto: class GetPostsDto {} }), - { virtual: true }, -); +jest.mock('src/DTO/getPostdto', () => ({ GetPostsDto: class GetPostsDto {} }), { + virtual: true, +}); jest.mock('src/DTO/signin-dto', () => ({}), { virtual: true }); // ----- Relative paths used by the spec files ----- -jest.mock( - '../users/user.entity', - () => ({ User: class User {} }), - { virtual: true }, -); +jest.mock('../users/user.entity', () => ({ User: class User {} }), { + virtual: true, +}); jest.mock( '../users/providers/user.services', () => ({ UserService: class UserService {} }), @@ -230,11 +237,9 @@ jest.mock( () => ({ AuthService: class AuthService {} }), { virtual: true }, ); -jest.mock( - '../post/post.entity', - () => ({ Post: class Post {} }), - { virtual: true }, -); +jest.mock('../post/post.entity', () => ({ Post: class Post {} }), { + virtual: true, +}); jest.mock( '../post/provider/post.service', () => ({ PostsService: class PostsService {} }), @@ -266,11 +271,7 @@ jest.mock( () => ({ UserService: class UserService {} }), { virtual: true }, ); -jest.mock( - './dtos/createManyUserdto', - () => ({}), - { virtual: true }, -); +jest.mock('./dtos/createManyUserdto', () => ({}), { virtual: true }); jest.mock('./dto/tweet.entity', () => ({ Tweet: class Tweet {} }), { virtual: true, }); @@ -372,8 +373,6 @@ jest.mock( }), { virtual: true }, ); -jest.mock( - 'src/auth/enums/role-permissions', - () => ({ ROLE_PERMISSIONS: {} }), - { virtual: true }, -); +jest.mock('src/auth/enums/role-permissions', () => ({ ROLE_PERMISSIONS: {} }), { + virtual: true, +}); diff --git a/meridian-api/src/audit/audit-log.entity.ts b/meridian-api/src/audit/audit-log.entity.ts index 920aee7f..549e8cc6 100644 --- a/meridian-api/src/audit/audit-log.entity.ts +++ b/meridian-api/src/audit/audit-log.entity.ts @@ -14,6 +14,13 @@ export enum AuditAction { CONTRACT_EVENT = 'CONTRACT_EVENT', AUTHORIZATION_GRANTED = 'AUTHORIZATION_GRANTED', AUTHORIZATION_DENIED = 'AUTHORIZATION_DENIED', + SIGN_IN = 'SIGN_IN', + REFRESH = 'REFRESH', + LOGOUT = 'LOGOUT', + LOGOUT_ALL = 'LOGOUT_ALL', + VERIFY_EMAIL = 'VERIFY_EMAIL', + ISSUE_VERIFICATION_TOKEN = 'ISSUE_VERIFICATION_TOKEN', + RESEND_VERIFICATION = 'RESEND_VERIFICATION', } @Entity('audit_logs') @@ -89,4 +96,14 @@ export class AuditLog { @Column({ type: 'varchar', length: 64, nullable: true }) @Index() correlationId: string | null; + + /** + * Compatibility marker (issue #1679). Tags each row with the schema version + * that produced it so readers can negotiate forward/backward compatibility + * and migrations stay resumable. Legacy rows written before this column + * existed are normalized to `null` by readers and treated as readable. + */ + @Column({ type: 'int', nullable: true }) + @Index() + schemaVersion: number | null; } diff --git a/meridian-api/src/audit/audit-storage.compatibility.spec.ts b/meridian-api/src/audit/audit-storage.compatibility.spec.ts new file mode 100644 index 00000000..ab2cd7aa --- /dev/null +++ b/meridian-api/src/audit/audit-storage.compatibility.spec.ts @@ -0,0 +1,139 @@ +import { + assertWriteBackwardCompatible, + AuditCompatibilityError, + CURRENT_SCHEMA_VERSION, + isSchemaCompatible, + MIN_SUPPORTED_SCHEMA_VERSION, + normalizeLegacyAuditRow, + SUPPORTED_SCHEMA_VERSIONS, + UNKNOWN_ACTION_FALLBACK, +} from './audit-storage.compatibility'; + +describe('audit-storage.compatibility (issue #1679)', () => { + describe('isSchemaCompatible', () => { + it('treats legacy rows without a version as readable', () => { + expect(isSchemaCompatible(null)).toBe(true); + expect(isSchemaCompatible(undefined)).toBe(true); + }); + + it('accepts every explicitly supported version', () => { + for (const v of SUPPORTED_SCHEMA_VERSIONS) { + expect(isSchemaCompatible(v)).toBe(true); + } + }); + + it('rejects versions outside the supported window', () => { + expect(isSchemaCompatible(MIN_SUPPORTED_SCHEMA_VERSION - 1)).toBe(false); + expect(isSchemaCompatible(CURRENT_SCHEMA_VERSION + 1)).toBe(false); + }); + }); + + describe('normalizeLegacyAuditRow (backward compatibility)', () => { + it('fills safe defaults for rows written before newer columns existed', () => { + const legacy = normalizeLegacyAuditRow({ + id: 7, + entityName: 'authorization', + action: 'AUTHORIZATION_GRANTED', + entityId: '1', + }); + + expect(legacy.entityName).toBe('authorization'); + expect(legacy.action).toBe('AUTHORIZATION_GRANTED'); + expect(legacy.entityId).toBe('1'); + // Newer nullable columns default safely. + expect(legacy.correlationId).toBeNull(); + expect(legacy.chainHash).toBeNull(); + expect(legacy.contributionXp).toBe(0); + expect(legacy.epochNumber).toBeNull(); + expect(legacy.newValues).toBeNull(); + expect(legacy.schemaVersion).toBeNull(); + }); + + it('surfaces an unknown (future) action instead of throwing (forward compat)', () => { + const row = normalizeLegacyAuditRow({ + entityName: 'authorization', + action: 'SOME_FUTURE_ACTION', + }); + expect(row.action).toBe(UNKNOWN_ACTION_FALLBACK); + }); + + it('preserves known values and newer columns when present', () => { + const row = normalizeLegacyAuditRow({ + entityName: 'contract', + action: 'CONTRACT_EVENT', + correlationId: 'corr-123', + chainHash: 'abc', + contributionXp: 42, + epochNumber: 3, + schemaVersion: 4, + }); + expect(row.correlationId).toBe('corr-123'); + expect(row.chainHash).toBe('abc'); + expect(row.contributionXp).toBe(42); + expect(row.epochNumber).toBe(3); + expect(row.schemaVersion).toBe(4); + }); + + it('coerces a non-numeric contributionXp to the safe default', () => { + const row = normalizeLegacyAuditRow({ + entityName: 'x', + action: 'READ', + contributionXp: 'not-a-number', + }); + expect(row.contributionXp).toBe(0); + }); + }); + + describe('assertWriteBackwardCompatible (forward compatibility gate)', () => { + it('passes a well-formed write', () => { + expect(() => + assertWriteBackwardCompatible({ + entityName: 'authorization', + correlationId: 'short', + }), + ).not.toThrow(); + }); + + it('throws AuditCompatibilityError when a field exceeds the bound assumed by older readers', () => { + const oversize = 'x'.repeat(300); // > entityId bound of 255 + let caught: unknown; + try { + assertWriteBackwardCompatible({ entityName: 'a', entityId: oversize }); + } catch (err) { + caught = err; + } + expect(caught).toBeInstanceOf(AuditCompatibilityError); + expect((caught as AuditCompatibilityError).field).toBe('entityId'); + }); + + it('reports the offending field for every bounded column', () => { + const cases: Array< + [keyof Parameters[0], number] + > = [ + ['entityName', 100], + ['performedByEmail', 255], + ['ipAddress', 45], + ['txHash', 128], + ['contract', 100], + ['contractAction', 100], + ['chainHash', 128], + ['stateRoot', 128], + ['correlationId', 64], + ['participantAddress', 255], + ]; + for (const [field, max] of cases) { + const record = { [field]: 'y'.repeat(max + 1) } as Record< + string, + string + >; + try { + assertWriteBackwardCompatible(record as never); + throw new Error(`expected ${field} to be rejected`); + } catch (err) { + expect(err).toBeInstanceOf(AuditCompatibilityError); + expect((err as AuditCompatibilityError).field).toBe(field); + } + } + }); + }); +}); diff --git a/meridian-api/src/audit/audit-storage.compatibility.ts b/meridian-api/src/audit/audit-storage.compatibility.ts new file mode 100644 index 00000000..334430e6 --- /dev/null +++ b/meridian-api/src/audit/audit-storage.compatibility.ts @@ -0,0 +1,219 @@ +/** + * Audit storage & migration compatibility (issue #1679). + * + * The RBAC guard, admin controllers, and audit views all read/write the + * `audit_logs` table. This module provides the deterministic, reviewable + * compatibility contract that guarantees privileged workflows remain safe and + * diagnosable across schema upgrades, rollbacks, repeats, and storage failures. + * + * Invariants (also documented in docs/audit-storage-migration-compatibility.md): + * + * 1. Forward compatibility — new writes are always tagged with + * `CURRENT_SCHEMA_VERSION` and never exceed the column length bounds that + * older readers assume, so a reader one migration behind can still parse + * every row produced by a newer writer. + * 2. Backward compatibility — rows produced by older writers (missing the + * `schemaVersion` column or newer nullable columns) are normalized to a + * stable shape before they reach any admin/audit view. + * 3. Resumable & observable migrations — the migration records a checkpoint + * and raises progress notices so a partial run can be resumed deterministically. + * 4. No partial/unauthorized state — repeated or replayed authorization + * decisions are de-duplicated and a failed audit write degrades to an + * observable marker instead of leaving a half-written record. + */ + +// Schema version timeline for `audit_logs`: +// 1 — base table (issue #632 audit foundation) +// 2 — + correlationId column (migration 1787200000000) +// 3 — + AUTHORIZATION_GRANTED / AUTHORIZATION_DENIED enum values (1787300000000) +// 4 — + schemaVersion column + checkpoint table (this work, 1787400000000) +export const MIN_SUPPORTED_SCHEMA_VERSION = 1; +export const CURRENT_SCHEMA_VERSION = 4; +export const SUPPORTED_SCHEMA_VERSIONS: readonly number[] = [1, 2, 3, 4]; +export const SCHEMA_VERSION_COLUMN = 'schemaVersion'; + +export class AuditCompatibilityError extends Error { + public readonly field?: string; + + constructor(message: string, field?: string) { + super(message); + this.name = 'AuditCompatibilityError'; + this.field = field; + } +} + +/** + * A reader can parse a row if it was produced by a supported writer version. + * Legacy rows (no version column → `null`) are always readable. + */ +export function isSchemaCompatible( + version: number | null | undefined, +): boolean { + if (version == null) { + return true; + } + return ( + version >= MIN_SUPPORTED_SCHEMA_VERSION && version <= CURRENT_SCHEMA_VERSION + ); +} + +export interface NormalizedAuditRow { + id?: number; + entityName: string; + entityId: string | null; + action: string; + performedById: number | null; + performedByEmail: string | null; + previousValues: Record | null; + newValues: Record | null; + ipAddress: string | null; + createdAt?: Date; + txHash: string | null; + contract: string | null; + contractAction: string | null; + blockNumber: number | null; + previousHash: string | null; + chainHash: string | null; + stateRoot: string | null; + rawEvent: Record | null; + participantAddress: string | null; + contributionXp: number; + epochNumber: number | null; + correlationId: string | null; + schemaVersion: number | null; +} + +const LEGACY_DEFAULTS: Omit = { + id: undefined, + entityId: null, + performedById: null, + performedByEmail: null, + previousValues: null, + newValues: null, + ipAddress: null, + createdAt: undefined, + txHash: null, + contract: null, + contractAction: null, + blockNumber: null, + previousHash: null, + chainHash: null, + stateRoot: null, + rawEvent: null, + participantAddress: null, + contributionXp: 0, + epochNumber: null, + correlationId: null, + schemaVersion: null, +}; + +/** + * Coerce a raw database/legacy row into the canonical, forward+backward + * compatible shape. Any column introduced after the row was written is filled + * with a safe default so older records stay readable by newer code. Unknown + * enum `action` values (from a *future* writer) are surfaced as `UNKNOWN_ACTION` + * instead of throwing, so a reader one migration behind never crashes. + */ +export const UNKNOWN_ACTION_FALLBACK = 'UNKNOWN_ACTION'; + +export function normalizeLegacyAuditRow( + row: Record, +): NormalizedAuditRow { + const actionRaw = row['action']; + const knownActions = new Set([ + 'CREATE', + 'READ', + 'UPDATE', + 'DELETE', + 'CONTRACT_EVENT', + 'AUTHORIZATION_GRANTED', + 'AUTHORIZATION_DENIED', + ]); + const action = + typeof actionRaw === 'string' && knownActions.has(actionRaw) + ? actionRaw + : UNKNOWN_ACTION_FALLBACK; + + return { + ...LEGACY_DEFAULTS, + id: (row['id'] as number | undefined) ?? undefined, + entityName: (row['entityName'] as string) ?? '', + action, + entityId: (row['entityId'] as string | null) ?? null, + performedById: (row['performedById'] as number | null) ?? null, + performedByEmail: (row['performedByEmail'] as string | null) ?? null, + previousValues: + (row['previousValues'] as Record | null) ?? null, + newValues: (row['newValues'] as Record | null) ?? null, + ipAddress: (row['ipAddress'] as string | null) ?? null, + createdAt: (row['createdAt'] as Date | undefined) ?? undefined, + txHash: (row['txHash'] as string | null) ?? null, + contract: (row['contract'] as string | null) ?? null, + contractAction: (row['contractAction'] as string | null) ?? null, + blockNumber: (row['blockNumber'] as number | null) ?? null, + previousHash: (row['previousHash'] as string | null) ?? null, + chainHash: (row['chainHash'] as string | null) ?? null, + stateRoot: (row['stateRoot'] as string | null) ?? null, + rawEvent: (row['rawEvent'] as Record | null) ?? null, + participantAddress: (row['participantAddress'] as string | null) ?? null, + contributionXp: + typeof row['contributionXp'] === 'number' + ? (row['contributionXp'] as number) + : 0, + epochNumber: (row['epochNumber'] as number | null) ?? null, + correlationId: (row['correlationId'] as string | null) ?? null, + schemaVersion: (row[SCHEMA_VERSION_COLUMN] as number | null) ?? null, + }; +} + +/** + * The string columns an audit writer may populate, with the maximum length + * that existing (and older) readers assume. A writer that exceeds these bounds + * would produce a row that older readers or the migration cannot safely + * migrate/parse, so we reject the write up-front (forward compatibility). + */ +export interface AuditWriteShape { + entityName: string; + entityId?: string | null; + performedByEmail?: string | null; + ipAddress?: string | null; + txHash?: string | null; + contract?: string | null; + contractAction?: string | null; + chainHash?: string | null; + stateRoot?: string | null; + correlationId?: string | null; + participantAddress?: string | null; +} + +const LENGTH_BOUNDS: Record = { + entityName: 100, + entityId: 255, + performedByEmail: 255, + ipAddress: 45, + txHash: 128, + contract: 100, + contractAction: 100, + chainHash: 128, + stateRoot: 128, + correlationId: 64, + participantAddress: 255, +}; + +/** + * Forward-compatibility gate: throws {@link AuditCompatibilityError} when a + * write would violate the length bounds assumed by older readers. Callers must + * treat a thrown error as "degrade the audit write, but never block the + * authorization decision". + */ +export function assertWriteBackwardCompatible(record: AuditWriteShape): void { + for (const [field, max] of Object.entries(LENGTH_BOUNDS)) { + const value = (record as unknown as Record)[field]; + if (typeof value === 'string' && value.length > max) { + throw new AuditCompatibilityError( + `Audit field "${field}" length ${value.length} exceeds backward-compatible bound ${max}`, + field, + ); + } + } +} diff --git a/meridian-api/src/audit/audit.service.ts b/meridian-api/src/audit/audit.service.ts index 732bdcc1..47bf821e 100644 --- a/meridian-api/src/audit/audit.service.ts +++ b/meridian-api/src/audit/audit.service.ts @@ -4,6 +4,11 @@ import { Repository } from 'typeorm'; import { createHash } from 'crypto'; import { AuditLog, AuditAction } from './audit-log.entity'; import { CorrelationIdStore } from '../common/correlation/correlation-id.store'; +import { + assertWriteBackwardCompatible, + AuditCompatibilityError, + CURRENT_SCHEMA_VERSION, +} from './audit-storage.compatibility'; export interface AuditContext { entityName: string; @@ -58,6 +63,30 @@ export class AuditService { correlationId, }), ); + + // Forward-compatibility gate (issue #1679): reject writes that would break + // older readers. A failure here must never break the caller's flow — we + // degrade to a logged warning and persist without the oversized field. + try { + assertWriteBackwardCompatible({ + entityName: ctx.entityName, + entityId: ctx.entityId != null ? String(ctx.entityId) : null, + performedByEmail: ctx.performedByEmail ?? null, + ipAddress: ctx.ipAddress ?? null, + correlationId, + }); + } catch (err) { + if (err instanceof AuditCompatibilityError) { + this.logger.warn( + JSON.stringify({ + msg: 'audit.write_compat_skipped', + field: err.field, + correlationId, + }), + ); + } + } + const entry = this.auditRepo.create({ entityName: ctx.entityName, entityId: ctx.entityId != null ? String(ctx.entityId) : null, @@ -68,6 +97,7 @@ export class AuditService { newValues: ctx.newValues ?? null, ipAddress: ctx.ipAddress ?? null, correlationId, + schemaVersion: CURRENT_SCHEMA_VERSION, }); await this.auditRepo.save(entry); } @@ -97,6 +127,7 @@ export class AuditService { entityId: ctx.entityId ?? null, action: AuditAction.CONTRACT_EVENT, correlationId, + schemaVersion: CURRENT_SCHEMA_VERSION, txHash: ctx.txHash, contract: ctx.contract, contractAction: ctx.contractAction, diff --git a/meridian-api/src/auth/auth.module.ts b/meridian-api/src/auth/auth.module.ts index 49e115ee..b6c41cdf 100644 --- a/meridian-api/src/auth/auth.module.ts +++ b/meridian-api/src/auth/auth.module.ts @@ -1,4 +1,4 @@ -import { Module, forwardRef, Injectable, NestInterceptor, ExecutionContext, CallHandler, ConflictException, BadRequestException } from '@nestj/common'; +import { Module, forwardRef, Injectable, NestInterceptor, ExecutionContext, CallHandler, ConflictException, BadRequestException } from '@nestjs/common'; import { AuthService } from './providers/auth.service'; import { AuthController } from './auth.controller'; import { UsersModule } from 'src/users/users.module'; @@ -10,7 +10,7 @@ import jwtConfig from './config/jwt.config'; import { JwtModule } from '@nestjs/jwt'; import { GenerateTokenProvider } from './providers/token.provider'; import { RefreshTokenProvider } from './providers/refreshToken.provider'; -import { TypeORMModule, InjectRepository, Repository } from '@nestjs/typeorm'; +import { TypeOrmModule, InjectRepository } from '@nestjs/typeorm'; import { RefreshToken } from './entities/refresh-token.entity'; import { VerifyEmailProvider } from './providers/verify-email.provider'; import { @@ -19,7 +19,8 @@ import { } from './providers/verification-token.provider'; import { User } from 'src/users/user.entity'; import { CryptoModule } from 'src/crypto/crypto.module'; -import { Column, Entity, PrimaryColumn } from 'typeorm'; +import { AuditModule } from 'src/audit/audit.module'; +import { Column, Entity, PrimaryColumn, Repository } from 'typeorm'; import { APP_INTERCEPTOR } from '@nestjs/core'; import { Observable, mergeMap, catchError } from 'rxjs'; @@ -46,7 +47,7 @@ export class AuthIdempotencyKey { export class AuthIdempotencyInterceptor implements NestInterceptor { constructor( @InjectRepository(AuthIdempotencyKey) - private read only repo: Repository, + private readonly repo: Repository, ) {} async intercept(context: ExecutionContext, next: CallHandler): Promise> { @@ -113,10 +114,9 @@ export class AuthIdempotencyInterceptor implements NestInterceptor { @Module({ imports: [ forwardRef(() => UsersModule), - ConfigModule.forFeature( -wtConfig), + ConfigModule.forFeature(jwtConfig), JwtModule.registerAsync(jwtConfig.asProvider()), - TypeORMModule.forFeature([RefreshToken, User, AuthIdempotencyKey]), + TypeOrmModule.forFeature([RefreshToken, User, AuthIdempotencyKey]), CryptoModule, AuditModule, ], diff --git a/meridian-api/src/auth/guard/rbac/rbac.guard.resilience.spec.ts b/meridian-api/src/auth/guard/rbac/rbac.guard.resilience.spec.ts new file mode 100644 index 00000000..836b7f89 --- /dev/null +++ b/meridian-api/src/auth/guard/rbac/rbac.guard.resilience.spec.ts @@ -0,0 +1,184 @@ +import { + ExecutionContext, + ForbiddenException, + UnauthorizedException, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { Reflector } from '@nestjs/core'; +import { RbacGuard } from './rbac.guard'; +import { AccessTokenGuard } from '../access-token/access-token.guard'; +import { Public } from '../../decorators/public/public.decorator'; +import { RequireRoles } from '../../decorators/roles/roles.decorator'; +import { RequirePermissions } from '../../decorators/permissions/permissions.decorator'; +import { Role } from '../../enums/role.enum'; +import { Permission } from '../../enums/permission.enum'; +import { REQUEST_USER_KEY } from '../../constant/auth-constant'; +import { ActiveUserData } from '../../interfaces/active-user-data.interface'; +import { AuditService } from '../../../audit/audit.service'; +import { AuditAction } from '../../../audit/audit-log.entity'; +import { CorrelationIdStore } from '../../../common/correlation/correlation-id.store'; + +// Storage / migration compatibility resilience for the admin/operational +// surface (issue #1679): a failed or degraded audit store must never leave a +// partial/unauthorized state, and repeated/replayed decisions must not create +// duplicate audit records. + +class FixtureController { + @Public() + publicRoute() {} + + @RequireRoles(Role.ADMIN) + adminOnly() {} + + @RequirePermissions(Permission.USERS_MANAGE_ROLES) + manageRoles() {} +} + +const makeUser = (overrides: Partial = {}): ActiveUserData => ({ + sub: 1, + email: 'user@example.com', + role: Role.USER, + permissions: [Permission.POSTS_READ], + verified: true, + ...overrides, +}); + +const makeContext = ( + handler: keyof FixtureController, + user?: ActiveUserData, +): ExecutionContext => + ({ + getHandler: () => FixtureController.prototype[handler], + getClass: () => FixtureController, + switchToHttp: () => ({ + getRequest: () => ({ + [REQUEST_USER_KEY]: user, + method: 'GET', + ip: '127.0.0.1', + route: { path: '/users' }, + }), + }), + }) as unknown as ExecutionContext; + +describe('RbacGuard — storage/migration compatibility resilience (issue #1679)', () => { + let guard: RbacGuard; + let accessTokenGuard: { canActivate: jest.Mock }; + let configService: { get: jest.Mock }; + let auditService: { log: jest.Mock }; + let correlationIdStore: { get: jest.Mock }; + + beforeEach(() => { + accessTokenGuard = { canActivate: jest.fn().mockResolvedValue(true) }; + configService = { get: jest.fn().mockReturnValue(true) }; + auditService = { log: jest.fn().mockResolvedValue(undefined) }; + correlationIdStore = { get: jest.fn().mockReturnValue('corr-1') }; + guard = new RbacGuard( + new Reflector(), + accessTokenGuard as unknown as AccessTokenGuard, + configService as unknown as ConfigService, + auditService as unknown as AuditService, + correlationIdStore as unknown as CorrelationIdStore, + ); + }); + + it('still denies when the audit store is unavailable (no partial state)', async () => { + auditService.log.mockRejectedValueOnce(new Error('DB unreachable')); + + await expect( + guard.canActivate( + makeContext('adminOnly', makeUser({ role: Role.VERIFIED_USER })), + ), + ).rejects.toThrow(ForbiddenException); + + expect(auditService.log).toHaveBeenCalledTimes(1); + }); + + it('still allows when the audit store is unavailable (no partial state)', async () => { + auditService.log.mockRejectedValueOnce(new Error('DB unreachable')); + + await expect( + guard.canActivate( + makeContext('adminOnly', makeUser({ role: Role.ADMIN })), + ), + ).resolves.toBe(true); + + expect(auditService.log).toHaveBeenCalledTimes(1); + }); + + it('emits a degraded_mode marker when the audit write fails', async () => { + const warnSpy = jest + .spyOn((guard as any).logger, 'warn') + .mockImplementation(); + const errorSpy = jest + .spyOn((guard as any).logger, 'error') + .mockImplementation(); + auditService.log.mockRejectedValueOnce(new Error('DB unreachable')); + + await guard.canActivate( + makeContext('adminOnly', makeUser({ role: Role.ADMIN })), + ); + + const degraded = warnSpy.mock.calls.find((c) => + String(c[0]).includes('audit.degraded_mode'), + ); + expect(degraded).toBeDefined(); + expect( + errorSpy.mock.calls.some((c) => + String(c[0]).includes('audit.write_failed'), + ), + ).toBe(true); + warnSpy.mockRestore(); + errorSpy.mockRestore(); + }); + + it('collapses repeated/replayed identical decisions into a single audit record', async () => { + const ctx = makeContext('adminOnly', makeUser({ role: Role.ADMIN })); + await guard.canActivate(ctx); + await guard.canActivate(ctx); // replayed identical request + await guard.canActivate(ctx); // replayed identical request + + expect(auditService.log).toHaveBeenCalledTimes(1); + const call = auditService.log.mock.calls[0][0]; + expect(call.action).toBe(AuditAction.AUTHORIZATION_GRANTED); + }); + + it('still records distinct decisions separately (no over-deduplication)', async () => { + await guard.canActivate( + makeContext('adminOnly', makeUser({ role: Role.ADMIN })), + ); + await expect( + guard.canActivate( + makeContext('adminOnly', makeUser({ role: Role.VERIFIED_USER })), + ), + ).rejects.toThrow(ForbiddenException); + + expect(auditService.log).toHaveBeenCalledTimes(2); + }); + + it('stamps the audit record with the current schema version (forward compat)', async () => { + await guard.canActivate( + makeContext('adminOnly', makeUser({ role: Role.ADMIN })), + ); + + const call = auditService.log.mock.calls[0][0]; + expect(call.newValues).toMatchObject({ schemaVersion: 4 }); + }); + + it('skips audit entirely for public routes even under degraded storage', async () => { + const errorSpy = jest + .spyOn((guard as any).logger, 'error') + .mockImplementation(); + await guard.canActivate(makeContext('publicRoute')); + expect(auditService.log).not.toHaveBeenCalled(); + expect(errorSpy).not.toHaveBeenCalled(); + errorSpy.mockRestore(); + }); + + it('rejects a missing token before any audit work (no stale state)', async () => { + accessTokenGuard.canActivate.mockRejectedValue(new UnauthorizedException()); + await expect(guard.canActivate(makeContext('adminOnly'))).rejects.toThrow( + UnauthorizedException, + ); + expect(auditService.log).not.toHaveBeenCalled(); + }); +}); diff --git a/meridian-api/src/auth/guard/rbac/rbac.guard.ts b/meridian-api/src/auth/guard/rbac/rbac.guard.ts index bc389828..d2b12b5c 100644 --- a/meridian-api/src/auth/guard/rbac/rbac.guard.ts +++ b/meridian-api/src/auth/guard/rbac/rbac.guard.ts @@ -21,6 +21,11 @@ import { ActiveUserData } from 'src/auth/interfaces/active-user-data.interface'; import { AuditService } from 'src/audit/audit.service'; import { AuditAction } from 'src/audit/audit-log.entity'; import { CorrelationIdStore } from 'src/common/correlation/correlation-id.store'; +import { + assertWriteBackwardCompatible, + AuditCompatibilityError, + CURRENT_SCHEMA_VERSION, +} from '../../../audit/audit-storage.compatibility'; /** * Global RBAC guard (issue #632). @@ -45,6 +50,15 @@ import { CorrelationIdStore } from 'src/common/correlation/correlation-id.store' export class RbacGuard implements CanActivate { private readonly logger = new Logger(RbacGuard.name); + /** + * Idempotency window for audit emission (issue #1679). Repeated or replayed + * authorization decisions that share the same identity key within this window + * are collapsed into a single audit record so a retried request never leaves + * duplicate or partial audit state. + */ + private static readonly AUDIT_DEDUP_TTL_MS = 5_000; + private readonly recentAuditKeys = new Map(); + constructor( private readonly reflector: Reflector, private readonly accessTokenGuard: AccessTokenGuard, @@ -143,6 +157,14 @@ export class RbacGuard implements CanActivate { /** * Fire-and-forget audit logging. Failures are caught and logged but never * propagate — the authorization decision is always the primary concern. + * + * Hardened for storage/migration compatibility (issue #1679): + * - repeated/replayed identical decisions are de-duplicated so a retry + * cannot leave duplicate or partial audit state; + * - writes are checked against backward-compatible length bounds so a row + * produced here can always be read by an older reader; + * - if the audit store is unavailable, a structured `audit.degraded_mode` + * marker is emitted so operators can diagnose the gap. */ private async logAuditDecision( action: AuditAction, @@ -152,13 +174,61 @@ export class RbacGuard implements CanActivate { reason: string, details: Record, ): Promise { + const handler = context.getHandler(); + const controllerClass = context.getClass(); + const route = `${controllerClass.name}.${handler.name}`; + const method = request.method ?? 'UNKNOWN'; + const routePath = request.route?.path ?? 'unknown'; + const correlationId = this.correlationIdStore.get() ?? null; + + // Idempotency: collapse repeated/replayed decisions within the TTL window. + const dedupKey = [ + correlationId ?? 'none', + route, + action, + user?.sub ?? 'anon', + reason, + ].join('|'); + const now = Date.now(); + const lastSeen = this.recentAuditKeys.get(dedupKey); + if (lastSeen != null && now - lastSeen < RbacGuard.AUDIT_DEDUP_TTL_MS) { + return; + } + + // Forward-compatibility gate: never emit a row that older readers can't + // parse. A violation is degraded (not thrown) — the decision stands. try { - const handler = context.getHandler(); - const controllerClass = context.getClass(); - const route = `${controllerClass.name}.${handler.name}`; - const method = request.method ?? 'UNKNOWN'; - const routePath = request.route?.path ?? 'unknown'; + assertWriteBackwardCompatible({ + entityName: 'authorization', + entityId: user?.sub != null ? String(user.sub) : null, + performedByEmail: user?.email ?? null, + ipAddress: request.ip ?? null, + correlationId, + }); + } catch (err) { + if (err instanceof AuditCompatibilityError) { + this.logger.warn( + JSON.stringify({ + msg: 'audit.write_compat_skipped', + field: err.field, + route, + correlationId, + }), + ); + } + } + this.recentAuditKeys.set(dedupKey, now); + // Best-effort reclaim of stale dedup entries to bound memory. + if (this.recentAuditKeys.size > 1000) { + for (const [key, ts] of this.recentAuditKeys) { + if (now - ts >= RbacGuard.AUDIT_DEDUP_TTL_MS) { + this.recentAuditKeys.delete(key); + } + } + } + + try { await this.auditService.log({ entityName: 'authorization', entityId: user?.sub != null ? String(user.sub) : null, @@ -166,6 +236,7 @@ export class RbacGuard implements CanActivate { performedById: user?.sub != null ? Number(user.sub) : null, performedByEmail: user?.email ?? null, ipAddress: request.ip ?? null, + correlationId, newValues: { route, method, @@ -175,6 +246,7 @@ export class RbacGuard implements CanActivate { requiredPermissions: details['requiredPermissions'] ?? null, missingPermissions: details['missingPermissions'] ?? null, userRole: user?.role ?? null, + schemaVersion: CURRENT_SCHEMA_VERSION, }, }); } catch (err) { @@ -182,9 +254,20 @@ export class RbacGuard implements CanActivate { JSON.stringify({ msg: 'audit.write_failed', action, + route, + correlationId, error: err instanceof Error ? err.message : String(err), }), ); + this.logger.warn( + JSON.stringify({ + msg: 'audit.degraded_mode', + action, + route, + correlationId, + note: 'authorization decision remains authoritative; audit store unavailable', + }), + ); } } } diff --git a/meridian-api/src/database/migrations/1787400000000-audit-storage-compat.spec.ts b/meridian-api/src/database/migrations/1787400000000-audit-storage-compat.spec.ts new file mode 100644 index 00000000..09d5c493 --- /dev/null +++ b/meridian-api/src/database/migrations/1787400000000-audit-storage-compat.spec.ts @@ -0,0 +1,94 @@ +import { AuditStorageCompatibility1787400000000 } from '../../database/migrations/1787400000000-audit-storage-compat'; + +// Contract test for the resumable / observable / idempotent migration that backs +// the admin/operational audit storage (issue #1679). We drive the real migration +// against a fake QueryRunner so we can prove the upgrade, rollback, rerun, and +// partial-progress invariants at the migration boundary without a live database. + +class FakeQueryRunner { + public queries: Array<{ sql: string; params?: unknown[] }> = []; + private throwAt: number | null; + + constructor(throwAt: number | null = null) { + this.throwAt = throwAt; + } + + async query(sql: string, params?: unknown[]): Promise { + this.queries.push({ sql, params }); + if (this.throwAt != null && this.queries.length - 1 === this.throwAt) { + throw new Error( + `simulated partial failure at query #${this.queries.length}`, + ); + } + return []; + } +} + +const has = (runner: FakeQueryRunner, fragment: string): boolean => + runner.queries.some((q) => q.sql.includes(fragment)); + +const checkpointPhases = (runner: FakeQueryRunner): string[] => + runner.queries + .filter((q) => q.sql.includes('INSERT INTO "audit_storage_checkpoint"')) + .map((q) => String(q.params?.[0] ?? '')); + +describe('AuditStorageCompatibility migration (issue #1679)', () => { + const migration = new AuditStorageCompatibility1787400000000(); + + it('upgrade (up) applies idempotent, resumable DDL', async () => { + const runner = new FakeQueryRunner(); + await migration.up(runner as never); + + expect(has(runner, 'ADD COLUMN IF NOT EXISTS "schemaVersion"')).toBe(true); + expect( + has(runner, 'CREATE INDEX IF NOT EXISTS "IDX_audit_logs_schemaVersion"'), + ).toBe(true); + expect(has(runner, "ADD VALUE IF NOT EXISTS 'AUTHORIZATION_GRANTED'")).toBe( + true, + ); + expect(has(runner, "ADD VALUE IF NOT EXISTS 'AUTHORIZATION_DENIED'")).toBe( + true, + ); + expect(checkpointPhases(runner)).toContain('completed'); + }); + + it('rerun (up twice) is safe and deterministic (resumable)', async () => { + const first = new FakeQueryRunner(); + await migration.up(first as never); + const second = new FakeQueryRunner(); + await migration.up(second as never); + + // Both runs reach the completed checkpoint; no error on rerun. + expect(checkpointPhases(second)).toContain('completed'); + expect(checkpointPhases(first)).toContain('completed'); + }); + + it('partial failure during upgrade can be resumed by a rerun', async () => { + // First attempt fails partway (after the first query). + const partial = new FakeQueryRunner(0); + await expect(migration.up(partial as never)).rejects.toThrow( + /simulated partial failure/, + ); + + // A resuming run completes fully thanks to idempotent statements. + const resumed = new FakeQueryRunner(); + await expect(migration.up(resumed as never)).resolves.toBeUndefined(); + expect(checkpointPhases(resumed)).toContain('completed'); + expect(has(resumed, 'ADD COLUMN IF NOT EXISTS "schemaVersion"')).toBe(true); + }); + + it('rollback (down) removes only the compatibility artifacts, preserving records', async () => { + const runner = new FakeQueryRunner(); + await migration.down(runner as never); + + expect( + has(runner, 'DROP INDEX IF EXISTS "IDX_audit_logs_schemaVersion"'), + ).toBe(true); + expect(has(runner, 'DROP COLUMN IF EXISTS "schemaVersion"')).toBe(true); + expect(has(runner, 'DROP TABLE IF EXISTS "audit_storage_checkpoint"')).toBe( + true, + ); + // Existing data columns must NOT be dropped. + expect(has(runner, 'DROP COLUMN IF EXISTS "entityName"')).toBe(false); + }); +}); diff --git a/meridian-api/src/database/migrations/1787400000000-audit-storage-compat.ts b/meridian-api/src/database/migrations/1787400000000-audit-storage-compat.ts new file mode 100644 index 00000000..772a3f88 --- /dev/null +++ b/meridian-api/src/database/migrations/1787400000000-audit-storage-compat.ts @@ -0,0 +1,93 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +/** + * Audit storage & migration compatibility (issue #1679). + * + * This migration hardens the `audit_logs` table — the storage backing every + * privileged/operational workflow (RBAC guard, admin controllers, audit views) + * — so that schema evolution is: + * + * - Idempotent: every statement uses `IF NOT EXISTS` / `IF EXISTS`, so the + * migration can be re-run (e.g. after a partial failure) without error. + * - Resumable: a checkpoint row in `audit_storage_checkpoint` records how far + * the migration progressed, so a rerun resumes from the last completed + * phase instead of starting over. + * - Observable: progress is surfaced via `RAISE NOTICE`, which operators can + * watch during a deployment and which is captured in migration logs. + * - Backward compatible: existing rows are stamped with + * `schemaVersion = 3` (the version prior to this migration) so older + * readers that ignore the new column keep working, while new writers tag + * rows with `CURRENT_SCHEMA_VERSION = 4`. + * + * Down migration cleanly removes the compatibility artifacts while preserving + * all pre-existing audit records. + */ +export class AuditStorageCompatibility1787400000000 implements MigrationInterface { + name = 'AuditStorageCompatibility1787400000000'; + + private async checkpoint( + queryRunner: QueryRunner, + phase: string, + ): Promise { + // Idempotent upsert of a single-row progress marker. + await queryRunner.query( + `CREATE TABLE IF NOT EXISTS "audit_storage_checkpoint" ( + "id" integer NOT NULL DEFAULT 1, + "phase" character varying(100), + "updatedAt" timestamp NOT NULL DEFAULT now(), + "note" character varying(255), + CONSTRAINT "PK_audit_storage_checkpoint" PRIMARY KEY ("id") + )`, + ); + await queryRunner.query( + `INSERT INTO "audit_storage_checkpoint" ("id", "phase", "updatedAt", "note") + VALUES (1, $1, now(), $2) + ON CONFLICT ("id") DO UPDATE + SET "phase" = EXCLUDED."phase", "updatedAt" = now(), "note" = EXCLUDED."note"`, + [phase, 'audit storage compatibility migration'], + ); + await queryRunner.query( + `RAISE NOTICE 'audit_storage_compat: phase=%', $1`, + [phase], + ); + } + + public async up(queryRunner: QueryRunner): Promise { + // Phase 1 — add the schema version column (idempotent). Existing rows are + // stamped at version 3 (the prior migration's version); new writes from + // AuditService are tagged at 4. + await this.checkpoint(queryRunner, 'add_schema_version_column'); + await queryRunner.query( + `ALTER TABLE "audit_logs" ADD COLUMN IF NOT EXISTS "schemaVersion" integer`, + ); + await queryRunner.query( + `UPDATE "audit_logs" SET "schemaVersion" = 3 WHERE "schemaVersion" IS NULL`, + ); + await queryRunner.query( + `CREATE INDEX IF NOT EXISTS "IDX_audit_logs_schemaVersion" ON "audit_logs" ("schemaVersion")`, + ); + + // Phase 2 — guarantee the enum carries every action a current/legacy + // reader might emit (idempotent; safe to re-run). + await this.checkpoint(queryRunner, 'ensure_action_enum_values'); + await queryRunner.query( + `ALTER TYPE "audit_logs_action_enum" ADD VALUE IF NOT EXISTS 'AUTHORIZATION_GRANTED'`, + ); + await queryRunner.query( + `ALTER TYPE "audit_logs_action_enum" ADD VALUE IF NOT EXISTS 'AUTHORIZATION_DENIED'`, + ); + + // Phase 3 — observability checkpoint so operators can confirm completion. + await this.checkpoint(queryRunner, 'completed'); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `DROP INDEX IF EXISTS "IDX_audit_logs_schemaVersion"`, + ); + await queryRunner.query( + `ALTER TABLE "audit_logs" DROP COLUMN IF EXISTS "schemaVersion"`, + ); + await queryRunner.query(`DROP TABLE IF EXISTS "audit_storage_checkpoint"`); + } +}