diff --git a/MQTT_DEVICE_AUTH.md b/MQTT_DEVICE_AUTH.md index 7b52837..a2eb168 100644 --- a/MQTT_DEVICE_AUTH.md +++ b/MQTT_DEVICE_AUTH.md @@ -8,14 +8,16 @@ describes from a usage/dashboard perspective. ## Overview The broker (`eclipse-mosquitto`, service `mqtt` / container `ul-hub-mqtt`) -exposes four listeners, each with a different trust model: +exposes three listeners, each with a different trust model: | Port | Protocol | Auth | Who it's for | |------|----------|------|---------------| | `1883` | MQTT (plaintext) | username/password | internal services (e.g. the Laravel subscriber, Python publisher) | -| `9001` (mapped to host `9005`) | MQTT over WebSocket | none (anonymous) | browser clients (dashboard) | | `8883` | MQTT over TLS | **client certificate** | field devices | -| `8884` | MQTT over TLS | username/password (server cert only) | administrators | +| `8884` | MQTT over TLS | username/password (server cert only) | scoped app clients and administrators | + +The former anonymous WebSocket listener on `9001` is disabled. The current web portal receives +live updates from the backend rather than connecting directly to Mosquitto. Device authentication happens on **port 8883**. A device presents a client certificate signed by the app's own Certificate Authority (CA); Mosquitto @@ -123,6 +125,10 @@ since they all share `acl_file /mosquitto/config/devices.acl`: # Certificate CN becomes the MQTT username — restrict each device to its own namespace. pattern readwrite devices/%u/# +# App username is orgId; app clients may publish durable points and read acks only in that org. +pattern write ul/%u/app/+/log/points +pattern read ul/%u/app/+/ack + # Admin: user topic readwrite # @@ -169,9 +175,9 @@ Managed via the same `/certificates` page: | Client | Listener | Auth | Can publish/subscribe | |---|---|---|---| | Field device (cert CN = serial) | 8883 (TLS) | client cert | `devices//#` only | +| App (username = orgId) | 8884 (TLS) | scoped per-org username/password | publish `ul//app/+/log/points`; read `ul//app/+/ack` | | Internal service (e.g. subscriber) | 1883 | username/password | depends on ACL entry for that username — none defined by default beyond `devices/%u/#`, so a plain username with no matching device row is effectively scoped to `devices//#` too, unless added as an admin | | Admin | 8884 (TLS) or 1883 | username/password | `#` and `$SYS/#` (everything) | -| Browser dashboard | 9001/9005 (WebSocket) | anonymous | no ACL applied — `allow_anonymous true`, so effectively unrestricted; treat this listener as untrusted/read-only in front-end code | ## Known gaps @@ -180,5 +186,5 @@ Managed via the same `/certificates` page: needs to be production-grade. - No automatic reload of `mosquitto.conf`/ACL/passwd changes — every provisioning action requires a manual `docker compose restart mqtt`. -- The WebSocket listener (9001) is fully anonymous with no ACL, so anything - reachable on port 9005 should be treated as public. +- Sprint 2's per-org app credential permits ack visibility across clients in the same org. This + accepted interim limitation is removed when OIDC-derived per-client broker identity lands. diff --git a/backend/prisma/migrations/20260821000000_app_ingest_idempotency/migration.sql b/backend/prisma/migrations/20260821000000_app_ingest_idempotency/migration.sql new file mode 100644 index 0000000..02b62f2 --- /dev/null +++ b/backend/prisma/migrations/20260821000000_app_ingest_idempotency/migration.sql @@ -0,0 +1,21 @@ +-- App-path ingest idempotency + provenance (Sprint 2, S2-a; SRS-SYN-2/7, §3.4.2). +-- Adds the UUIDv7 idempotency key and forward-compat provenance tags to locate_points. + +-- CreateEnum +CREATE TYPE "PointOrigin" AS ENUM ('APP', 'LOCATOR', 'MAGLINK'); + +-- CreateEnum +CREATE TYPE "PointUploadPath" AS ENUM ('APP_MQTT', 'DEVICE_MQTT', 'REST_BATCH'); + +-- AlterTable +ALTER TABLE "locate_points" ADD COLUMN "pointId" TEXT, +ADD COLUMN "origin" "PointOrigin" NOT NULL DEFAULT 'LOCATOR', +ADD COLUMN "uploadPath" "PointUploadPath", +ADD COLUMN "originClientId" TEXT, +ADD COLUMN "createdAt" TIMESTAMPTZ(6); + +-- CreateIndex +CREATE UNIQUE INDEX "locate_points_pointId_key" ON "locate_points"("pointId"); + +-- CreateIndex +CREATE INDEX "locate_points_jobId_createdAt_idx" ON "locate_points"("jobId", "createdAt"); diff --git a/backend/prisma/schema.prisma b/backend/prisma/schema.prisma index 68d12fa..6c1d578 100644 --- a/backend/prisma/schema.prisma +++ b/backend/prisma/schema.prisma @@ -53,6 +53,22 @@ enum LocateMode { SONDE } +// Which producer a point originated from (telemetry-schema.md `origin`). Tagged now +// for SYN-6 path arbitration; no arbitration LOGIC runs yet (deferred, sprint-2 scope). +enum PointOrigin { + APP + LOCATOR + MAGLINK +} + +// Which transport carried the point to the cloud (telemetry-schema.md `uploadPath`). +// Same forward-compat tagging as PointOrigin. +enum PointUploadPath { + APP_MQTT + DEVICE_MQTT + REST_BATCH +} + model Organization { id String @id @default(cuid()) name String @@ -185,6 +201,25 @@ model LocatePoint { id BigInt @id @default(autoincrement()) jobId String deviceId String? + + // Client-generated UUIDv7, the idempotency key across every upload path + // (app MQTT, device-direct MQTT, REST batch — SRS §3.4.2 / telemetry-schema.md). + // A replayed UUID is ingested exactly once (unique constraint below). Nullable: + // the legacy device-direct path (LogIngest/PointsIngest) predates it and does not + // supply one yet; the app path (SYN-2) always does. + pointId String? @unique + + // Provenance tags (telemetry-schema.md). Set on the app path now; carried for + // forward-compat with SYN-6 arbitration, which does not run yet. + origin PointOrigin @default(LOCATOR) + uploadPath PointUploadPath? + originClientId String? + + // Event time asserted by the producer (record creation, telemetry-schema.md + // `createdAt`). Ingest orders and dedups by this, NOT by arrival (`receivedAt`), + // so replays and out-of-order delivery converge to the same stored ordering. + createdAt DateTime? @db.Timestamptz(6) + lat Decimal @db.Decimal(10, 8) lng Decimal @db.Decimal(11, 8) altitude Decimal? @db.Decimal(8, 3) @@ -217,7 +252,11 @@ model LocatePoint { job Job @relation(fields: [jobId], references: [id], onDelete: Cascade) device Device? @relation(fields: [deviceId], references: [id], onDelete: SetNull) + // (jobId, recordedAt) serves GPS-time reads; (jobId, createdAt) serves the + // event-time ordering the app path and portal read use (SRS-SYN-2). pointId is + // already uniquely indexed above — that index also backs the dedup lookup. @@index([jobId, recordedAt]) + @@index([jobId, createdAt]) @@map("locate_points") } diff --git a/backend/src/ingest/app-log-ingest.service.spec.ts b/backend/src/ingest/app-log-ingest.service.spec.ts new file mode 100644 index 0000000..d3fe76c --- /dev/null +++ b/backend/src/ingest/app-log-ingest.service.spec.ts @@ -0,0 +1,315 @@ +import { Test } from '@nestjs/testing'; +import { readFileSync } from 'fs'; +import { resolve } from 'path'; +import { PrismaService } from '../prisma/prisma.service'; +import { RealtimeService } from '../realtime/realtime.service'; +import { AppLogAck } from './dto/app-messages.dto'; +import { AppLogIngestService } from './app-log-ingest.service'; +import { MqttClientService } from './mqtt-client.service'; + +// Unit coverage for the durable app-log ingest path (S2-a, SRS-SYN-2/7, §3.4.2). +// Prisma is an in-memory fake keyed on pointId so idempotency is exercised for real +// (a replayed UUID must be stored exactly once), without a database. The broker is a +// spy so we can assert the application-level ack shape and topic. +describe('AppLogIngestService', () => { + const ORG = 'org_alpha'; + const CLIENT = 'app_client_1'; + const JOB = { id: 'job_1', orgId: ORG, ticketNumber: 'T-1' }; + + let service: AppLogIngestService; + let published: { topic: string; payload: AppLogAck }[]; + let store: Map; + + const point = ( + pointId: string, + createdAt: string, + extra: Record = {}, + ) => ({ + pointId, + createdAt, + origin: 'APP', + uploadPath: 'APP_MQTT', + lat: 40, + lng: -80, + ts: createdAt, + qualityFlag: 'IN_SPEC', + ...extra, + }); + + const msg = ( + points: object[], + job: { jobId?: string; ticket?: string } = { jobId: JOB.id }, + ) => JSON.stringify({ schemaVersion: '1', ...job, points }); + + const lastAck = () => published[published.length - 1].payload; + + beforeEach(async () => { + published = []; + store = new Map(); + + const prisma = { + organization: { + findUnique: jest.fn(({ where }: { where: { id: string } }) => + Promise.resolve(where.id === ORG ? { id: ORG } : null), + ), + }, + job: { + findFirst: jest.fn( + ({ where }: { where: { id: string; orgId: string } }) => + Promise.resolve( + where.id === JOB.id && where.orgId === ORG ? JOB : null, + ), + ), + findUnique: jest.fn(() => Promise.resolve(null)), + create: jest.fn(() => Promise.resolve(JOB)), + }, + locatePoint: { + findMany: jest.fn( + ({ where }: { where: { pointId: { in: string[] } } }) => + Promise.resolve( + where.pointId.in + .filter((id) => store.has(id)) + .map((pointId) => ({ pointId })), + ), + ), + createManyAndReturn: jest.fn( + ({ + data, + }: { + data: { pointId: string; jobId: string; createdAt: Date }[]; + }) => { + const inserted: unknown[] = []; + for (const row of data) { + if (store.has(row.pointId)) { + continue; // skipDuplicates + } + store.set(row.pointId, row); + inserted.push({ + ...row, + id: BigInt(store.size), + lat: 40, + lng: -80, + altitude: null, + utilityType: 'UNKNOWN', + sequence: null, + fixType: 'NONE', + hAccuracy: null, + vAccuracy: null, + satellites: null, + hdop: null, + depth: null, + frequencyHz: null, + currentMa: null, + signalDb: null, + gainDb: null, + locateMode: null, + phaseDeg: null, + compassDeg: null, + distortionPct: null, + recordedAt: row.createdAt, + receivedAt: row.createdAt, + origin: 'APP', + uploadPath: 'APP_MQTT', + originClientId: CLIENT, + deviceId: null, + }); + } + return Promise.resolve(inserted); + }, + ), + }, + }; + + const mqtt = { + publish: jest.fn((topic: string, payload: AppLogAck) => + published.push({ topic, payload }), + ), + }; + + const moduleRef = await Test.createTestingModule({ + providers: [ + AppLogIngestService, + { provide: PrismaService, useValue: prisma }, + { provide: RealtimeService, useValue: { publish: jest.fn() } }, + { provide: MqttClientService, useValue: mqtt }, + ], + }).compile(); + + service = moduleRef.get(AppLogIngestService); + }); + + it('ingests a fresh point once and acks it accepted on the client namespace', async () => { + await service.handle( + ORG, + CLIENT, + msg([ + point('018f1a00-0000-7000-8000-000000000001', '2026-08-21T10:00:00Z'), + ]), + ); + + expect(store.size).toBe(1); + expect(published[0].topic).toBe(`ul/${ORG}/app/${CLIENT}/ack`); + const ack = lastAck(); + expect(ack.results).toEqual([ + { pointId: '018f1a00-0000-7000-8000-000000000001', outcome: 'ACCEPTED' }, + ]); + }); + + it('consumes the shared app fixture and emits the shared ack fixture', async () => { + const fixtureRoot = resolve( + __dirname, + '../../../../meta/contracts/fixtures', + ); + const publishFixture = readFileSync( + resolve(fixtureRoot, 'app-log-points-v1.json'), + 'utf8', + ); + const expectedAck = JSON.parse( + readFileSync(resolve(fixtureRoot, 'app-log-ack-v1.json'), 'utf8'), + ); + + await service.handle(ORG, CLIENT, publishFixture); + + expect(store.size).toBe(1); + expect(lastAck()).toEqual(expectedAck); + }); + + it('ingests a replayed UUID exactly once (idempotency) and acks it as duplicate', async () => { + const p = point( + '018f1a00-0000-7000-8000-000000000002', + '2026-08-21T10:00:00Z', + ); + await service.handle(ORG, CLIENT, msg([p])); + await service.handle(ORG, CLIENT, msg([p])); // replay + + expect(store.size).toBe(1); // stored exactly once + const ack = lastAck(); + expect(ack.results).toEqual([ + { pointId: '018f1a00-0000-7000-8000-000000000002', outcome: 'DUPLICATE' }, + ]); + }); + + it('orders a batch by event time, not array order', async () => { + const later = point( + '018f1a00-0000-7000-8000-00000000000a', + '2026-08-21T10:05:00Z', + ); + const earlier = point( + '018f1a00-0000-7000-8000-00000000000b', + '2026-08-21T10:01:00Z', + ); + await service.handle(ORG, CLIENT, msg([later, earlier])); + + const rows = [...store.values()]; + expect(rows.map((r) => r.pointId)).toEqual([ + earlier.pointId, + later.pointId, + ]); + }); + + it('rejects a point whose jobId belongs to another org (cross-namespace backstop)', async () => { + await service.handle( + ORG, + CLIENT, + msg( + [point('018f1a00-0000-7000-8000-00000000000c', '2026-08-21T10:00:00Z')], + { jobId: 'job_of_org_beta' }, + ), + ); + + expect(store.size).toBe(0); + const ack = lastAck(); + expect(ack.results).toEqual([ + { + pointId: '018f1a00-0000-7000-8000-00000000000c', + outcome: 'REJECTED', + reasonCode: 'UNKNOWN_JOB_OR_WRONG_ORG', + }, + ]); + }); + + it('drops (no ack) a message on an unknown org namespace', async () => { + await service.handle( + 'org_ghost', + CLIENT, + msg([ + point('018f1a00-0000-7000-8000-00000000000d', '2026-08-21T10:00:00Z'), + ]), + ); + expect(published).toHaveLength(0); + expect(store.size).toBe(0); + }); + + it('does not emit an ambiguous ack when malformed JSON has no usable pointId', async () => { + await service.handle(ORG, CLIENT, '{ not json'); + expect(published).toHaveLength(0); + }); + + it('rejects a schema-invalid point but names its UUID in the ack', async () => { + // Missing required createdAt + bad lat → validation fails; pointId still surfaced. + await service.handle( + ORG, + CLIENT, + msg([ + { + pointId: '018f1a00-0000-7000-8000-00000000000e', + lat: 999, + lng: 0, + ts: '2026-08-21T10:00:00Z', + }, + ]), + ); + const ack = lastAck(); + expect(store.size).toBe(0); + expect(ack.results).toEqual([ + { + pointId: '018f1a00-0000-7000-8000-00000000000e', + outcome: 'REJECTED', + reasonCode: 'VALIDATION_ERROR', + }, + ]); + }); + + it('accepts a valid point while rejecting a malformed neighbor in the same batch', async () => { + const bad = point( + '018f1a00-0000-7000-8000-00000000000f', + '2026-08-21T10:00:00Z', + { lat: 999 }, + ); + const good = point( + '018f1a00-0000-7000-8000-000000000010', + '2026-08-21T10:00:01Z', + ); + await service.handle(ORG, CLIENT, msg([bad, good])); + + expect(store.size).toBe(1); + expect(lastAck().results).toEqual( + expect.arrayContaining([ + { + pointId: bad.pointId, + outcome: 'REJECTED', + reasonCode: 'VALIDATION_ERROR', + }, + { pointId: good.pointId, outcome: 'ACCEPTED' }, + ]), + ); + }); + + it('rejects a non-v7 UUID even when it is otherwise a valid UUID', async () => { + const uuidV4 = '550e8400-e29b-41d4-a716-446655440000'; + await service.handle( + ORG, + CLIENT, + msg([point(uuidV4, '2026-08-21T10:00:00Z')]), + ); + + expect(store.size).toBe(0); + expect(lastAck().results).toEqual([ + { + pointId: uuidV4, + outcome: 'REJECTED', + reasonCode: 'POINT_ID_NOT_UUIDV7', + }, + ]); + }); +}); diff --git a/backend/src/ingest/app-log-ingest.service.ts b/backend/src/ingest/app-log-ingest.service.ts new file mode 100644 index 0000000..2ce25e3 --- /dev/null +++ b/backend/src/ingest/app-log-ingest.service.ts @@ -0,0 +1,277 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { GpsFixType, Prisma } from '@prisma/client'; +import { plainToInstance } from 'class-transformer'; +import { validate } from 'class-validator'; +import { toPointDto } from '../points/points.service'; +import { PrismaService } from '../prisma/prisma.service'; +import { RealtimeService } from '../realtime/realtime.service'; +import { + AppFixType, + AppLogAck, + AppLogAckResult, + AppLogPointDto, + AppLogPointsMessageDto, +} from './dto/app-messages.dto'; +import { MqttClientService } from './mqtt-client.service'; + +const SCHEMA_VERSION = '1'; +const UUID_V7 = + /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +// Durable app-originated ingest for ul/{orgId}/app/{clientId}/log/points. The exact v1 +// publish/ack envelopes are frozen in meta/contracts/telemetry-schema.md. +@Injectable() +export class AppLogIngestService { + private readonly logger = new Logger(AppLogIngestService.name); + + constructor( + private readonly prisma: PrismaService, + private readonly realtime: RealtimeService, + private readonly mqttClient: MqttClientService, + ) {} + + async handle( + orgId: string, + clientId: string, + rawPayload: string, + ): Promise { + const org = await this.prisma.organization.findUnique({ + where: { id: orgId }, + select: { id: true }, + }); + if (!org) { + this.logger.warn( + `app-log for unknown org "${orgId}" (client ${clientId}) — dropped`, + ); + return; + } + + let parsed: unknown; + try { + parsed = JSON.parse(rawPayload); + } catch { + // There is no trustworthy pointId to acknowledge. Raw app payloads are redacted by the + // router; retain/log on the app side rather than sending a null/ambiguous acknowledgement. + this.logger.warn( + `invalid JSON from app ${orgId}/${clientId} — no acknowledgement emitted`, + ); + return; + } + + const msg = plainToInstance(AppLogPointsMessageDto, parsed as object); + const rawPoints = Array.isArray((parsed as { points?: unknown[] })?.points) + ? (parsed as { points: unknown[] }).points + : []; + const usableIds = rawPoints + .map((value) => + typeof (value as { pointId?: unknown })?.pointId === 'string' + ? (value as { pointId: string }).pointId + : null, + ) + .filter((value): value is string => value !== null); + + const hasJobId = typeof msg.jobId === 'string' && msg.jobId.length > 0; + const hasTicket = typeof msg.ticket === 'string' && msg.ticket.length > 0; + const envelopeValid = + msg.schemaVersion === SCHEMA_VERSION && + rawPoints.length > 0 && + rawPoints.length <= 500 && + hasJobId !== hasTicket; + if (!envelopeValid) { + this.publishAck( + orgId, + clientId, + usableIds.map((pointId) => ({ + pointId, + outcome: 'REJECTED', + reasonCode: 'ENVELOPE_VALIDATION_ERROR', + })), + ); + return; + } + + // Validate independently so one malformed capture does not discard valid captures in the + // same batch (LOG-7). Invalid UUID strings can still be named in a rejection; missing IDs + // cannot be safely correlated and therefore remain unacknowledged. + const results: AppLogAckResult[] = []; + const candidates: AppLogPointDto[] = []; + for (const rawPoint of rawPoints) { + const point = plainToInstance(AppLogPointDto, rawPoint as object); + const pointId = typeof point.pointId === 'string' ? point.pointId : null; + const errors = await validate(point, { whitelist: true }); + if (!pointId) { + continue; + } + if (errors.length > 0 || !UUID_V7.test(pointId)) { + results.push({ + pointId, + outcome: 'REJECTED', + reasonCode: UUID_V7.test(pointId) + ? 'VALIDATION_ERROR' + : 'POINT_ID_NOT_UUIDV7', + }); + continue; + } + candidates.push(point); + } + + const job = await this.resolveJob(orgId, clientId, msg); + if (!job) { + results.push( + ...candidates.map((point) => ({ + pointId: point.pointId, + outcome: 'REJECTED' as const, + reasonCode: 'UNKNOWN_JOB_OR_WRONG_ORG', + })), + ); + this.publishAck(orgId, clientId, results); + return; + } + + // Collapse duplicates inside this delivery, then use producer event time rather than + // arrival/array order. The first valid occurrence wins; repeated IDs are safe to release. + const byPointId = new Map(); + for (const point of candidates) { + if (byPointId.has(point.pointId)) { + results.push({ + pointId: point.pointId, + outcome: 'DUPLICATE', + reasonCode: 'DUPLICATE_IN_BATCH', + }); + } else { + byPointId.set(point.pointId, point); + } + } + const ordered = [...byPointId.values()].sort( + (a, b) => +new Date(a.createdAt) - +new Date(b.createdAt), + ); + + const seen = ordered.length + ? await this.prisma.locatePoint.findMany({ + where: { pointId: { in: ordered.map((point) => point.pointId) } }, + select: { pointId: true }, + }) + : []; + const seenIds = new Set(seen.map((row) => row.pointId)); + const fresh = ordered.filter((point) => !seenIds.has(point.pointId)); + for (const point of ordered.filter((candidate) => + seenIds.has(candidate.pointId), + )) { + results.push({ pointId: point.pointId, outcome: 'DUPLICATE' }); + } + + let stored: Awaited< + ReturnType + > = []; + if (fresh.length > 0) { + stored = await this.prisma.locatePoint.createManyAndReturn({ + data: fresh.map((point) => ({ + jobId: job.id, + deviceId: null, + pointId: point.pointId, + origin: 'APP' as const, + uploadPath: 'APP_MQTT' as const, + originClientId: clientId, + createdAt: new Date(point.createdAt), + lat: point.lat, + lng: point.lng, + altitude: point.alt, + utilityType: point.utility, + sequence: point.seq, + fixType: this.toStoredFix(point.fix), + hAccuracy: point.hAcc, + vAccuracy: point.vAcc, + satellites: point.sats, + hdop: point.hdop, + depth: point.depth, + frequencyHz: point.freqHz, + currentMa: point.currentMa, + signalDb: point.signalDb, + gainDb: point.gainDb, + locateMode: point.mode, + phaseDeg: point.phaseDeg, + compassDeg: point.compassDeg, + distortionPct: point.distortionPct, + recordedAt: new Date(point.ts), + raw: point as unknown as Prisma.InputJsonValue, + })), + skipDuplicates: true, + }); + } + + const storedIds = new Set(stored.map((row) => row.pointId)); + for (const point of fresh) { + results.push({ + pointId: point.pointId, + outcome: storedIds.has(point.pointId) ? 'ACCEPTED' : 'DUPLICATE', + }); + } + + if (stored.length > 0) { + this.realtime.publish(`job:${job.id}`, { + type: 'points', + jobId: job.id, + points: stored.map(toPointDto), + }); + } + this.publishAck(orgId, clientId, results); + } + + private async resolveJob( + orgId: string, + clientId: string, + msg: AppLogPointsMessageDto, + ) { + if (msg.jobId) { + const job = await this.prisma.job.findFirst({ + where: { id: msg.jobId, orgId }, + }); + if (!job) + this.logger.warn( + `app-log ${orgId}/${clientId} referenced a foreign/unknown job ${msg.jobId}`, + ); + return job; + } + const existing = await this.prisma.job.findUnique({ + where: { orgId_ticketNumber: { orgId, ticketNumber: msg.ticket! } }, + }); + if (existing) return existing; + return this.prisma.job.create({ + data: { + orgId, + ticketNumber: msg.ticket!, + title: `Ticket ${msg.ticket} (app-created)`, + status: 'IN_PROGRESS', + source: 'WEB', + startedAt: new Date(), + }, + }); + } + + private toStoredFix(fix?: AppFixType): GpsFixType | undefined { + switch (fix) { + case 'FLOAT': + return GpsFixType.FLOAT_RTK; + case 'FIXED': + return GpsFixType.FIXED_RTK; + case 'NO_FIX': + return GpsFixType.NONE; + case 'AUTONOMOUS': + return GpsFixType.AUTONOMOUS; + case 'DGPS': + return GpsFixType.DGPS; + default: + return undefined; + } + } + + private publishAck( + orgId: string, + clientId: string, + results: AppLogAckResult[], + ): void { + if (results.length === 0) return; + const ack: AppLogAck = { schemaVersion: SCHEMA_VERSION, results }; + this.mqttClient.publish(`ul/${orgId}/app/${clientId}/ack`, ack); + } +} diff --git a/backend/src/ingest/dto/app-messages.dto.ts b/backend/src/ingest/dto/app-messages.dto.ts new file mode 100644 index 0000000..d6a4b2c --- /dev/null +++ b/backend/src/ingest/dto/app-messages.dto.ts @@ -0,0 +1,189 @@ +import { LocateMode, UtilityType } from '@prisma/client'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { Type } from 'class-transformer'; +import { + ArrayMaxSize, + ArrayMinSize, + IsArray, + IsDateString, + IsEnum, + IsIn, + IsInt, + IsNumber, + IsOptional, + IsString, + IsUUID, + Max, + MaxLength, + Min, + ValidateNested, +} from 'class-validator'; + +export const APP_FIX_TYPES = [ + 'AUTONOMOUS', + 'DGPS', + 'FLOAT', + 'FIXED', + 'NO_FIX', +] as const; +export type AppFixType = (typeof APP_FIX_TYPES)[number]; + +// Frozen app MQTT wire profile v1 from meta/contracts/telemetry-schema.md. This is deliberately +// separate from MqttPointDto: the app wire uses the normative fix tokens FLOAT/FIXED/NO_FIX, +// while the legacy device path uses Prisma's FLOAT_RTK/FIXED_RTK/NONE storage tokens. +export class AppLogPointDto { + @ApiProperty({ + format: 'uuid', + description: 'Client-generated UUIDv7 idempotency key.', + }) + @IsUUID() + pointId: string; + + @ApiProperty({ format: 'date-time' }) + @IsDateString() + createdAt: string; + + @IsIn(['APP']) + origin: 'APP'; + + @IsIn(['APP_MQTT']) + uploadPath: 'APP_MQTT'; + + @IsNumber() + @Min(-90) + @Max(90) + lat: number; + + @IsNumber() + @Min(-180) + @Max(180) + lng: number; + + @IsOptional() + @IsNumber() + alt?: number; + + @IsDateString() + ts: string; + + @IsOptional() + @IsIn(APP_FIX_TYPES) + fix?: AppFixType; + + @IsOptional() + @IsNumber() + @Min(0) + hAcc?: number; + + @IsOptional() + @IsNumber() + @Min(0) + vAcc?: number; + + @IsOptional() + @IsInt() + @Min(0) + sats?: number; + + @IsOptional() + @IsNumber() + @Min(0) + hdop?: number; + + @IsOptional() + @IsNumber() + @Min(0) + depth?: number; + + @IsOptional() + @IsInt() + @Min(0) + freqHz?: number; + + @IsOptional() + @IsNumber() + @Min(0) + currentMa?: number; + + @IsOptional() + @IsNumber() + signalDb?: number; + + @IsOptional() + @IsNumber() + gainDb?: number; + + @IsOptional() + @IsEnum(LocateMode) + mode?: LocateMode; + + @IsOptional() + @IsNumber() + phaseDeg?: number; + + @IsOptional() + @IsNumber() + @Min(0) + @Max(360) + compassDeg?: number; + + @IsOptional() + @IsNumber() + @Min(0) + @Max(100) + distortionPct?: number; + + @IsOptional() + @IsEnum(UtilityType) + utility?: UtilityType; + + @IsOptional() + @IsInt() + seq?: number; + + @IsIn(['IN_SPEC', 'OUT_OF_SPEC', 'MANUAL', 'PPK_CORRECTED', 'WAIVED', 'NONCOMPLIANT']) + qualityFlag: 'IN_SPEC' | 'OUT_OF_SPEC' | 'MANUAL' | 'PPK_CORRECTED' | 'WAIVED' | 'NONCOMPLIANT'; +} + +export class AppLogPointsMessageDto { + @ApiPropertyOptional({ description: 'MQTT app-log wire profile version.' }) + @IsString() + @MaxLength(16) + schemaVersion: string; + + @ApiPropertyOptional({ + description: 'Job id; exactly one of jobId or ticket is required.', + }) + @IsOptional() + @IsString() + jobId?: string; + + @ApiPropertyOptional({ + description: 'Ticket number; exactly one of jobId or ticket is required.', + }) + @IsOptional() + @IsString() + @MaxLength(64) + ticket?: string; + + @ApiProperty({ type: [AppLogPointDto] }) + @IsArray() + @ArrayMinSize(1) + @ArrayMaxSize(500) + @ValidateNested({ each: true }) + @Type(() => AppLogPointDto) + points: AppLogPointDto[]; +} + +export type AppLogAckOutcome = 'ACCEPTED' | 'DUPLICATE' | 'REJECTED'; + +export interface AppLogAckResult { + pointId: string; + outcome: AppLogAckOutcome; + reasonCode?: string; +} + +export interface AppLogAck { + schemaVersion: string; + results: AppLogAckResult[]; +} diff --git a/backend/src/ingest/ingest-router.service.ts b/backend/src/ingest/ingest-router.service.ts index 6b3f8da..6a36258 100644 --- a/backend/src/ingest/ingest-router.service.ts +++ b/backend/src/ingest/ingest-router.service.ts @@ -1,5 +1,6 @@ import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { PrismaService } from '../prisma/prisma.service'; +import { AppLogIngestService } from './app-log-ingest.service'; import { JobsIngestService } from './jobs-ingest.service'; import { LogIngestService } from './log-ingest.service'; import { MqttClientService } from './mqtt-client.service'; @@ -15,6 +16,7 @@ export class IngestRouterService implements OnModuleInit { private readonly pointsIngest: PointsIngestService, private readonly jobsIngest: JobsIngestService, private readonly logIngest: LogIngestService, + private readonly appLogIngest: AppLogIngestService, ) {} onModuleInit() { @@ -26,10 +28,32 @@ export class IngestRouterService implements OnModuleInit { } private async route(topic: string, payload: string) { - // Raw log captures every message on devices/#, matched or not (audit/debug trail) - await this.prisma.deviceEvent.create({ data: { topic, payload } }); + const segments = topic.split('/'); + const root = segments[0]; - const [root, idSegment, ...rest] = topic.split('/'); + // App point payloads contain personal/location data. Preserve routing evidence without + // creating an unmanaged second copy of the telemetry in device_events (S2 security H3). + const auditPayload = + root === 'ul' && + segments[2] === 'app' && + segments.slice(4).join('/') === 'log/points' + ? JSON.stringify({ + redacted: true, + bytes: Buffer.byteLength(payload, 'utf8'), + }) + : payload; + await this.prisma.deviceEvent.create({ + data: { topic, payload: auditPayload }, + }); + + // App / device MQTT namespace: ul/{orgId}/{clientClass}/{clientId}/{subtopic...} + // (SRS §3.4.2). Sprint 2 handles the durable app-log path; other clientClasses and + // subtopics are represented but not yet ingested here. + if (root === 'ul') { + return this.routeUl(segments, payload); + } + + const [, idSegment, ...rest] = segments; const subtopic = rest.join('/'); if (root !== 'devices' || !idSegment || subtopic === 'jobs/ack') { return; // not device traffic, or our own ack echoed back @@ -42,13 +66,19 @@ export class IngestRouterService implements OnModuleInit { return; } - const device = await this.prisma.device.findUnique({ where: { mqttUsername: idSegment } }); + const device = await this.prisma.device.findUnique({ + where: { mqttUsername: idSegment }, + }); if (!device) { - this.logger.warn(`Message from unregistered device username "${idSegment}" (raw-logged only)`); + this.logger.warn( + `Message from unregistered device username "${idSegment}" (raw-logged only)`, + ); return; } if (!device.isActive) { - this.logger.warn(`Message from deactivated device "${idSegment}" ignored`); + this.logger.warn( + `Message from deactivated device "${idSegment}" ignored`, + ); return; } @@ -69,4 +99,29 @@ export class IngestRouterService implements OnModuleInit { this.logger.debug(`Unhandled subtopic "${subtopic}" from ${idSegment}`); } } + + // ul/{orgId}/{clientClass}/{clientId}/{subtopic...} — SRS §3.4.2 topic scheme. + // The namespace itself carries tenant (orgId) + publisher (clientId) identity; the + // broker ACL confines a publisher to its own namespace, and AppLogIngest re-checks + // tenant scope on the payload's job as a backstop. Only the durable app-log path is + // ingested this sprint. + private async routeUl(segments: string[], payload: string) { + const [, orgId, clientClass, clientId, ...rest] = segments; + const subtopic = rest.join('/'); + if (!orgId || !clientClass || !clientId) { + this.logger.warn(`Malformed ul topic "${segments.join('/')}" — ignored`); + return; + } + // Our own cloud→publisher acks are echoed back to us on the same subscription; skip. + if (subtopic === 'ack') { + return; + } + if (clientClass === 'app' && subtopic === 'log/points') { + await this.appLogIngest.handle(orgId, clientId, payload); + return; + } + this.logger.debug( + `Unhandled ul path "${clientClass}/${subtopic}" (org ${orgId}, client ${clientId})`, + ); + } } diff --git a/backend/src/ingest/ingest.module.ts b/backend/src/ingest/ingest.module.ts index fa06dfd..e9429a3 100644 --- a/backend/src/ingest/ingest.module.ts +++ b/backend/src/ingest/ingest.module.ts @@ -1,5 +1,6 @@ import { Module } from '@nestjs/common'; import { RealtimeModule } from '../realtime/realtime.module'; +import { AppLogIngestService } from './app-log-ingest.service'; import { IngestRouterService } from './ingest-router.service'; import { JobsIngestService } from './jobs-ingest.service'; import { LocatorRegistryService } from './locator-registry.service'; @@ -15,6 +16,7 @@ import { PointsIngestService } from './points-ingest.service'; PointsIngestService, JobsIngestService, LogIngestService, + AppLogIngestService, LocatorRegistryService, ], exports: [MqttClientService], diff --git a/backend/src/ingest/mqtt-client.service.ts b/backend/src/ingest/mqtt-client.service.ts index 15ce044..95ca98d 100644 --- a/backend/src/ingest/mqtt-client.service.ts +++ b/backend/src/ingest/mqtt-client.service.ts @@ -1,4 +1,9 @@ -import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { + Injectable, + Logger, + OnModuleDestroy, + OnModuleInit, +} from '@nestjs/common'; import { connect, MqttClient } from 'mqtt'; export type MqttMessageListener = (topic: string, payload: Buffer) => void; @@ -12,19 +17,29 @@ export class MqttClientService implements OnModuleInit, OnModuleDestroy { onModuleInit() { const host = process.env.MQTT_HOST || 'mosquitto'; const port = Number(process.env.MQTT_PORT || 1883); + const username = process.env.MQTT_USERNAME; + const password = process.env.MQTT_PASSWORD; + if (!username || !password) { + throw new Error( + 'MQTT_USERNAME and MQTT_PASSWORD are required; refusing to use a default broker credential.', + ); + } this.client = connect(`mqtt://${host}:${port}`, { - username: process.env.MQTT_USERNAME || 'backend', - password: process.env.MQTT_PASSWORD || 'backendpass', + username, + password, clientId: `ulhub-backend-${Math.random().toString(16).slice(2)}`, }); this.client.on('connect', () => { this.logger.log(`Connected to MQTT broker at ${host}:${port}`); - this.client!.subscribe('devices/#', (err) => { + // devices/# = legacy device-direct namespace; ul/# = SRS §3.4.2 topic scheme + // (app + device-direct). QoS 1 so the broker redelivers durable log records the + // backend missed while disconnected (broker loss ≠ capture loss, SRS-SYN-1/7). + this.client!.subscribe(['devices/#', 'ul/#'], { qos: 1 }, (err) => { if (err) { - this.logger.error('Failed to subscribe to devices/#', err); + this.logger.error('Failed to subscribe to devices/# + ul/#', err); } else { - this.logger.log('Subscribed to devices/#'); + this.logger.log('Subscribed to devices/# and ul/#'); } }); }); diff --git a/backend/src/points/points.service.ts b/backend/src/points/points.service.ts index 3c21eb0..e3a6331 100644 --- a/backend/src/points/points.service.ts +++ b/backend/src/points/points.service.ts @@ -6,8 +6,13 @@ import { CreatePointDto, QueryPointsDto } from './dto/points.dto'; // JSON-safe shape: BigInt id -> string, Decimal -> number export interface PointDto { id: string; + pointId: string | null; jobId: string; deviceId: string | null; + origin: string; + uploadPath: string | null; + originClientId: string | null; + createdAt: Date | null; lat: number; lng: number; altitude: number | null; @@ -38,8 +43,13 @@ function num(value: unknown): number | null { export function toPointDto(p: LocatePoint): PointDto { return { id: p.id.toString(), + pointId: p.pointId, jobId: p.jobId, deviceId: p.deviceId, + origin: p.origin, + uploadPath: p.uploadPath, + originClientId: p.originClientId, + createdAt: p.createdAt, lat: Number(p.lat), lng: Number(p.lng), altitude: num(p.altitude), diff --git a/docker-compose.yml b/docker-compose.yml index 219501e..3bc7196 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,9 +13,12 @@ services: ports: - "1883:1883" - "8883:8883" - - "9001:9001" + - "8884:8884" volumes: - ./mosquitto:/mosquitto + # Publicly trusted server identity for app MQTTS. The device listener still + # validates client certificates against /mosquitto/certs/ca.crt. + - /etc/letsencrypt:/etc/letsencrypt:ro backend: build: @@ -37,7 +40,7 @@ services: MQTT_HOST: mosquitto MQTT_PORT: 1883 MQTT_USERNAME: ${MQTT_BACKEND_USERNAME:-backend} - MQTT_PASSWORD: ${MQTT_BACKEND_PASSWORD:-backendpass} + MQTT_PASSWORD: ${MQTT_BACKEND_PASSWORD:?set MQTT_BACKEND_PASSWORD} MQTT_CERTS_DIR: /mosquitto-certs MQTT_TLS_PORT: 8883 # Must match the CN the broker's server cert was provisioned for @@ -81,4 +84,4 @@ services: - postgres volumes: - postgres-data: \ No newline at end of file + postgres-data: diff --git a/mosquitto/config/devices.acl b/mosquitto/config/devices.acl index e0a991b..58fbec6 100644 --- a/mosquitto/config/devices.acl +++ b/mosquitto/config/devices.acl @@ -12,10 +12,22 @@ topic readwrite # topic readwrite $SYS/# # Backend service: reads all device traffic, writes job acks back to devices, -# and publishes on behalf of the /sim simulator tool (devices//log) +# and publishes on behalf of the /sim simulator tool (devices//log). +# Also ingests the SRS §3.4.2 ul/# namespace (app + device-direct) and writes the +# application-level acks back on ul/{orgId}/{clientClass}/{clientId}/ack. user backend topic read devices/# topic write devices/+/jobs/ack topic write devices/+/log +topic read ul/# +topic write ul/+/+/+/ack + +# --- App-client namespace confinement (S2-a / SRS §3.4.2 Topic scheme) -------------- +# The interim app credential uses username == orgId, so %u confines it to that tenant. +# The '+' clientId wildcard is the accepted Sprint 2 tradeoff: clients within one org share +# the credential and can see that org's ack topics until OIDC-derived per-client identity lands. +# Publish/read permissions are narrower than readwrite '#': only durable point input and acks. +pattern write ul/%u/app/+/log/points +pattern read ul/%u/app/+/ack # testuser is a demo *device*: only the per-device pattern rule above applies diff --git a/mosquitto/config/mosquitto.conf b/mosquitto/config/mosquitto.conf index 7b899c7..a508bdd 100644 --- a/mosquitto/config/mosquitto.conf +++ b/mosquitto/config/mosquitto.conf @@ -6,11 +6,6 @@ password_file /mosquitto/config/passwd acl_file /mosquitto/config/devices.acl allow_anonymous false -# WebSocket — browser clients, no authentication required -listener 9001 0.0.0.0 -protocol websockets -allow_anonymous true - # TLS MQTT — devices authenticate with client certificates (port 8883) # require_certificate true forces client cert; cert CN becomes the MQTT username. # ACL restricts each device to devices//# @@ -19,20 +14,19 @@ allow_anonymous true # since there's no config/cert hot-reload. listener 8883 0.0.0.0 cafile /mosquitto/certs/ca.crt -certfile /mosquitto/certs/server.crt -keyfile /mosquitto/certs/server.key +certfile /etc/letsencrypt/live/dev.hub.umagul.net/fullchain.pem +keyfile /etc/letsencrypt/live/dev.hub.umagul.net/privkey.pem require_certificate true use_identity_as_username true allow_anonymous false acl_file /mosquitto/config/devices.acl -# TLS MQTT — admin access via username/password, no client cert required (port 8884) -# Connect with CA cert for server verification, then username/password. -# listener 8884 0.0.0.0 -# cafile /mosquitto/certs/ca.crt -# certfile /mosquitto/certs/server.crt -# keyfile /mosquitto/certs/server.key -# require_certificate false -# password_file /mosquitto/config/passwd -# allow_anonymous false -# acl_file /mosquitto/config/devices.acl +# TLS MQTT — app/admin username+password access (port 8884). App usernames are orgIds; +# devices.acl confines them to ul/{orgId}/app/... . No anonymous listener is exposed. +listener 8884 0.0.0.0 +certfile /etc/letsencrypt/live/dev.hub.umagul.net/fullchain.pem +keyfile /etc/letsencrypt/live/dev.hub.umagul.net/privkey.pem +require_certificate false +password_file /mosquitto/config/passwd +allow_anonymous false +acl_file /mosquitto/config/devices.acl diff --git a/nginx/dev.hub.umagul.net.conf b/nginx/dev.hub.umagul.net.conf index 29e3ba0..f49a5b8 100644 --- a/nginx/dev.hub.umagul.net.conf +++ b/nginx/dev.hub.umagul.net.conf @@ -6,18 +6,6 @@ server { root /var/www/certbot; } - location /mqtt { - # proxy MQTT over WebSocket (strip the /mqtt prefix so Mosquitto sees /) - proxy_pass http://127.0.0.1:9001/; - proxy_http_version 1.1; - proxy_set_header Upgrade $http_upgrade; - proxy_set_header Connection "upgrade"; - proxy_set_header Host $host; - proxy_set_header X-Real-IP $remote_addr; - proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; - proxy_read_timeout 300s; - } - location / { return 301 https://$host$request_uri; } @@ -33,14 +21,8 @@ server { client_max_body_size 20m; location /mqtt { - proxy_pass http://127.0.0.1:9001/; - proxy_http_version 1.1; - proxy_set_header Upgrade $http_upgrade; - proxy_set_header Connection "upgrade"; - proxy_set_header Host $host; - proxy_set_header X-Real-IP $remote_addr; - proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; - proxy_read_timeout 300s; + # Anonymous MQTT-over-WebSocket was removed for Sprint 2 security criterion B1. + return 410; } location / {