Compare commits

..

7 Commits

Author SHA1 Message Date
Brent Perteet
e552ae4e5c fix(S2-a): mount broker auth files outside source tree 2026-08-20 15:53:20 -05:00
Brent Perteet
8aea87c3cd fix(S2-a): keep broker credential database outside Git 2026-08-20 15:52:14 -05:00
Brent Perteet
8f638bfa4f fix(S2-a): keep plaintext MQTT inside compose network 2026-08-20 15:44:38 -05:00
Brent Perteet
9ffe021354 fix(S2-a): expose scoped app MQTT through WSS 2026-08-20 15:43:42 -05:00
Brent Perteet
b66ae2cc47 fix(S2-a): load public MQTT TLS cert from broker volume 2026-08-20 15:27:03 -05:00
Brent Perteet
8bcc12ec96 feat(S2-a): ingest durable app MQTT points securely 2026-08-20 15:22:40 -05:00
Brent Perteet
daa3407b9d test: wire jest harness for QA gate (S1-f)
Add jest + ts-jest to the NestJS backend with an app.service smoke spec that
exercises Nest DI. `npm test` is the documented command the QA gate runs.

Trace: SRS §6 gate, NFR-8.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-20 12:53:36 -05:00
18 changed files with 4340 additions and 293 deletions

View File

@@ -8,14 +8,31 @@ describes from a usage/dashboard perspective.
## Overview ## Overview
The broker (`eclipse-mosquitto`, service `mqtt` / container `ul-hub-mqtt`) 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 | | Port | Protocol | Auth | Who it's for |
|------|----------|------|---------------| |------|----------|------|---------------|
| `1883` | MQTT (plaintext) | username/password | internal services (e.g. the Laravel subscriber, Python publisher) | | `1883` | MQTT (plaintext, Compose network only) | username/password | internal backend service; not host-published |
| `9001` (mapped to host `9005`) | MQTT over WebSocket | none (anonymous) | browser clients (dashboard) |
| `8883` | MQTT over TLS | **client certificate** | field devices | | `8883` | MQTT over TLS | **client certificate** | field devices |
| `8884` | MQTT over TLS | username/password (server cert only) | administrators | | `443` (`/mqtt` → loopback `9001`) | MQTT over WSS/TLS | username/password | scoped app clients |
| `8884` | MQTT over TLS | username/password (server cert only) | scoped app clients and administrators on networks that expose the raw port |
The former anonymous WebSocket listener on `9001` is now authenticated, uses the same ACL as the
TLS listeners, and is bound to host loopback only. Nginx exposes it as WSS at `/mqtt` on port 443;
the web portal itself continues to receive live updates from the backend.
The TLS listeners use `mosquitto/certs/public-fullchain.pem` and
`mosquitto/certs/public-privkey.pem`, copied from the host's Let's Encrypt certificate during
deployment with owner `1883:1883` and mode `0600`. These files are deployment secrets/artifacts
and are excluded from Git.
The live password database is likewise outside Git at
`/home/ubuntu/.config/ul-platform/mosquitto.passwd`, bind-mounted read-only as
`/run/secrets/mosquitto_passwd`. The deployed ACL is copied to
`/home/ubuntu/.config/ul-platform/mosquitto.acl` and bind-mounted beside it. Both files are owned
by broker uid/gid 1883 with mode 0600. Provisioning updates the host password file and restarts Mosquitto; the
tracked `mosquitto/config/passwd` is only a legacy/bootstrap sample and must not receive new
organization credentials.
Device authentication happens on **port 8883**. A device presents a client Device authentication happens on **port 8883**. A device presents a client
certificate signed by the app's own Certificate Authority (CA); Mosquitto certificate signed by the app's own Certificate Authority (CA); Mosquitto
@@ -123,6 +140,10 @@ since they all share `acl_file /mosquitto/config/devices.acl`:
# Certificate CN becomes the MQTT username — restrict each device to its own namespace. # Certificate CN becomes the MQTT username — restrict each device to its own namespace.
pattern readwrite devices/%u/# 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: <username> # Admin: <username>
user <username> user <username>
topic readwrite # topic readwrite #
@@ -169,9 +190,9 @@ Managed via the same `/certificates` page:
| Client | Listener | Auth | Can publish/subscribe | | Client | Listener | Auth | Can publish/subscribe |
|---|---|---|---| |---|---|---|---|
| Field device (cert CN = serial) | 8883 (TLS) | client cert | `devices/<serial>/#` only | | Field device (cert CN = serial) | 8883 (TLS) | client cert | `devices/<serial>/#` only |
| App (username = orgId) | 8884 (TLS) | scoped per-org username/password | publish `ul/<orgId>/app/+/log/points`; read `ul/<orgId>/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/<username>/#` too, unless added as an admin | | 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/<username>/#` too, unless added as an admin |
| Admin | 8884 (TLS) or 1883 | username/password | `#` and `$SYS/#` (everything) | | 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 ## Known gaps
@@ -180,5 +201,5 @@ Managed via the same `/certificates` page:
needs to be production-grade. needs to be production-grade.
- No automatic reload of `mosquitto.conf`/ACL/passwd changes — every - No automatic reload of `mosquitto.conf`/ACL/passwd changes — every
provisioning action requires a manual `docker compose restart mqtt`. provisioning action requires a manual `docker compose restart mqtt`.
- The WebSocket listener (9001) is fully anonymous with no ACL, so anything - Sprint 2's per-org app credential permits ack visibility across clients in the same org. This
reachable on port 9005 should be treated as public. accepted interim limitation is removed when OIDC-derived per-client broker identity lands.

16
backend/jest.config.js Normal file
View File

@@ -0,0 +1,16 @@
/**
* Jest config for the UlHub backend (NestJS).
* ts-jest transpiles the TypeScript sources; decorators/metadata are honored via tsconfig.
* Specs live next to sources as *.spec.ts (unit) — e2e would go under test/ with a separate config.
*/
module.exports = {
moduleFileExtensions: ['js', 'json', 'ts'],
rootDir: 'src',
testRegex: '.*\\.spec\\.ts$',
transform: {
'^.+\\.ts$': ['ts-jest', { tsconfig: '<rootDir>/../tsconfig.json' }],
},
collectCoverageFrom: ['**/*.(t|j)s'],
coverageDirectory: '../coverage',
testEnvironment: 'node',
};

3527
backend/package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@@ -8,7 +8,9 @@
"build": "nest build", "build": "nest build",
"format": "prettier --write \"src/**/*.ts\"", "format": "prettier --write \"src/**/*.ts\"",
"postinstall": "prisma generate", "postinstall": "prisma generate",
"db:seed": "ts-node prisma/seed.ts" "db:seed": "ts-node prisma/seed.ts",
"test": "jest",
"test:ci": "jest --ci --runInBand"
}, },
"prisma": { "prisma": {
"seed": "ts-node prisma/seed.ts" "seed": "ts-node prisma/seed.ts"
@@ -44,9 +46,12 @@
"@types/cookie-parser": "^1.4.7", "@types/cookie-parser": "^1.4.7",
"@types/express": "^4.17.17", "@types/express": "^4.17.17",
"@types/node": "^20.11.0", "@types/node": "^20.11.0",
"@types/jest": "^29.5.12",
"@types/passport-jwt": "^4.0.1", "@types/passport-jwt": "^4.0.1",
"@types/ws": "^8.5.10", "@types/ws": "^8.5.10",
"jest": "^29.7.0",
"prettier": "^3.0.0", "prettier": "^3.0.0",
"ts-jest": "^29.2.5",
"prisma": "^6.10.0", "prisma": "^6.10.0",
"ts-node": "^10.9.1", "ts-node": "^10.9.1",
"typescript": "^5.5.0" "typescript": "^5.5.0"

View File

@@ -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");

View File

@@ -53,6 +53,22 @@ enum LocateMode {
SONDE 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 { model Organization {
id String @id @default(cuid()) id String @id @default(cuid())
name String name String
@@ -185,6 +201,25 @@ model LocatePoint {
id BigInt @id @default(autoincrement()) id BigInt @id @default(autoincrement())
jobId String jobId String
deviceId 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) lat Decimal @db.Decimal(10, 8)
lng Decimal @db.Decimal(11, 8) lng Decimal @db.Decimal(11, 8)
altitude Decimal? @db.Decimal(8, 3) altitude Decimal? @db.Decimal(8, 3)
@@ -217,7 +252,11 @@ model LocatePoint {
job Job @relation(fields: [jobId], references: [id], onDelete: Cascade) job Job @relation(fields: [jobId], references: [id], onDelete: Cascade)
device Device? @relation(fields: [deviceId], references: [id], onDelete: SetNull) 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, recordedAt])
@@index([jobId, createdAt])
@@map("locate_points") @@map("locate_points")
} }

View File

@@ -0,0 +1,27 @@
import { Test, TestingModule } from '@nestjs/testing';
import { AppService } from './app.service';
// Smoke test for the QA gate (S1-f, trace: SRS §6 gate / NFR-8).
// Exercises Nest DI wiring end-to-end for a pure service so a green run proves the
// toolchain (ts-jest + @nestjs/testing) is functional. Real coverage grows from here.
describe('AppService (smoke)', () => {
let service: AppService;
beforeEach(async () => {
const moduleRef: TestingModule = await Test.createTestingModule({
providers: [AppService],
}).compile();
service = moduleRef.get<AppService>(AppService);
});
it('is resolvable from the DI container', () => {
expect(service).toBeDefined();
});
it('returns the API welcome payload', () => {
expect(service.getHello()).toEqual({
message: 'Welcome to UlHub API',
docs: 'GET /api',
});
});
});

View File

@@ -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<string, { pointId: string; jobId: string; createdAt: Date }>;
const point = (
pointId: string,
createdAt: string,
extra: Record<string, unknown> = {},
) => ({
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',
},
]);
});
});

View File

@@ -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<void> {
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<string, AppLogPointDto>();
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<typeof this.prisma.locatePoint.createManyAndReturn>
> = [];
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);
}
}

View File

@@ -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[];
}

View File

@@ -1,5 +1,6 @@
import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { PrismaService } from '../prisma/prisma.service'; import { PrismaService } from '../prisma/prisma.service';
import { AppLogIngestService } from './app-log-ingest.service';
import { JobsIngestService } from './jobs-ingest.service'; import { JobsIngestService } from './jobs-ingest.service';
import { LogIngestService } from './log-ingest.service'; import { LogIngestService } from './log-ingest.service';
import { MqttClientService } from './mqtt-client.service'; import { MqttClientService } from './mqtt-client.service';
@@ -15,6 +16,7 @@ export class IngestRouterService implements OnModuleInit {
private readonly pointsIngest: PointsIngestService, private readonly pointsIngest: PointsIngestService,
private readonly jobsIngest: JobsIngestService, private readonly jobsIngest: JobsIngestService,
private readonly logIngest: LogIngestService, private readonly logIngest: LogIngestService,
private readonly appLogIngest: AppLogIngestService,
) {} ) {}
onModuleInit() { onModuleInit() {
@@ -26,10 +28,32 @@ export class IngestRouterService implements OnModuleInit {
} }
private async route(topic: string, payload: string) { private async route(topic: string, payload: string) {
// Raw log captures every message on devices/#, matched or not (audit/debug trail) const segments = topic.split('/');
await this.prisma.deviceEvent.create({ data: { topic, payload } }); 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('/'); const subtopic = rest.join('/');
if (root !== 'devices' || !idSegment || subtopic === 'jobs/ack') { if (root !== 'devices' || !idSegment || subtopic === 'jobs/ack') {
return; // not device traffic, or our own ack echoed back return; // not device traffic, or our own ack echoed back
@@ -42,13 +66,19 @@ export class IngestRouterService implements OnModuleInit {
return; return;
} }
const device = await this.prisma.device.findUnique({ where: { mqttUsername: idSegment } }); const device = await this.prisma.device.findUnique({
where: { mqttUsername: idSegment },
});
if (!device) { 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; return;
} }
if (!device.isActive) { if (!device.isActive) {
this.logger.warn(`Message from deactivated device "${idSegment}" ignored`); this.logger.warn(
`Message from deactivated device "${idSegment}" ignored`,
);
return; return;
} }
@@ -69,4 +99,29 @@ export class IngestRouterService implements OnModuleInit {
this.logger.debug(`Unhandled subtopic "${subtopic}" from ${idSegment}`); 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})`,
);
}
} }

View File

@@ -1,5 +1,6 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { RealtimeModule } from '../realtime/realtime.module'; import { RealtimeModule } from '../realtime/realtime.module';
import { AppLogIngestService } from './app-log-ingest.service';
import { IngestRouterService } from './ingest-router.service'; import { IngestRouterService } from './ingest-router.service';
import { JobsIngestService } from './jobs-ingest.service'; import { JobsIngestService } from './jobs-ingest.service';
import { LocatorRegistryService } from './locator-registry.service'; import { LocatorRegistryService } from './locator-registry.service';
@@ -15,6 +16,7 @@ import { PointsIngestService } from './points-ingest.service';
PointsIngestService, PointsIngestService,
JobsIngestService, JobsIngestService,
LogIngestService, LogIngestService,
AppLogIngestService,
LocatorRegistryService, LocatorRegistryService,
], ],
exports: [MqttClientService], exports: [MqttClientService],

View File

@@ -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'; import { connect, MqttClient } from 'mqtt';
export type MqttMessageListener = (topic: string, payload: Buffer) => void; export type MqttMessageListener = (topic: string, payload: Buffer) => void;
@@ -12,19 +17,29 @@ export class MqttClientService implements OnModuleInit, OnModuleDestroy {
onModuleInit() { onModuleInit() {
const host = process.env.MQTT_HOST || 'mosquitto'; const host = process.env.MQTT_HOST || 'mosquitto';
const port = Number(process.env.MQTT_PORT || 1883); 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}`, { this.client = connect(`mqtt://${host}:${port}`, {
username: process.env.MQTT_USERNAME || 'backend', username,
password: process.env.MQTT_PASSWORD || 'backendpass', password,
clientId: `ulhub-backend-${Math.random().toString(16).slice(2)}`, clientId: `ulhub-backend-${Math.random().toString(16).slice(2)}`,
}); });
this.client.on('connect', () => { this.client.on('connect', () => {
this.logger.log(`Connected to MQTT broker at ${host}:${port}`); 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) { if (err) {
this.logger.error('Failed to subscribe to devices/#', err); this.logger.error('Failed to subscribe to devices/# + ul/#', err);
} else { } else {
this.logger.log('Subscribed to devices/#'); this.logger.log('Subscribed to devices/# and ul/#');
} }
}); });
}); });

View File

@@ -6,8 +6,13 @@ import { CreatePointDto, QueryPointsDto } from './dto/points.dto';
// JSON-safe shape: BigInt id -> string, Decimal -> number // JSON-safe shape: BigInt id -> string, Decimal -> number
export interface PointDto { export interface PointDto {
id: string; id: string;
pointId: string | null;
jobId: string; jobId: string;
deviceId: string | null; deviceId: string | null;
origin: string;
uploadPath: string | null;
originClientId: string | null;
createdAt: Date | null;
lat: number; lat: number;
lng: number; lng: number;
altitude: number | null; altitude: number | null;
@@ -38,8 +43,13 @@ function num(value: unknown): number | null {
export function toPointDto(p: LocatePoint): PointDto { export function toPointDto(p: LocatePoint): PointDto {
return { return {
id: p.id.toString(), id: p.id.toString(),
pointId: p.pointId,
jobId: p.jobId, jobId: p.jobId,
deviceId: p.deviceId, deviceId: p.deviceId,
origin: p.origin,
uploadPath: p.uploadPath,
originClientId: p.originClientId,
createdAt: p.createdAt,
lat: Number(p.lat), lat: Number(p.lat),
lng: Number(p.lng), lng: Number(p.lng),
altitude: num(p.altitude), altitude: num(p.altitude),

View File

@@ -11,11 +11,13 @@ services:
mosquitto: mosquitto:
image: eclipse-mosquitto:2 image: eclipse-mosquitto:2
ports: ports:
- "1883:1883"
- "8883:8883" - "8883:8883"
- "9001:9001" - "8884:8884"
- "127.0.0.1:9001:9001"
volumes: volumes:
- ./mosquitto:/mosquitto - ./mosquitto:/mosquitto
- /home/ubuntu/.config/ul-platform/mosquitto.passwd:/run/secrets/mosquitto_passwd:ro
- /home/ubuntu/.config/ul-platform/mosquitto.acl:/run/secrets/mosquitto_acl:ro
backend: backend:
build: build:
@@ -37,7 +39,7 @@ services:
MQTT_HOST: mosquitto MQTT_HOST: mosquitto
MQTT_PORT: 1883 MQTT_PORT: 1883
MQTT_USERNAME: ${MQTT_BACKEND_USERNAME:-backend} 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_CERTS_DIR: /mosquitto-certs
MQTT_TLS_PORT: 8883 MQTT_TLS_PORT: 8883
# Must match the CN the broker's server cert was provisioned for # Must match the CN the broker's server cert was provisioned for

View File

@@ -12,10 +12,22 @@ topic readwrite #
topic readwrite $SYS/# topic readwrite $SYS/#
# Backend service: reads all device traffic, writes job acks back to devices, # Backend service: reads all device traffic, writes job acks back to devices,
# and publishes on behalf of the /sim simulator tool (devices/<serial>/log) # and publishes on behalf of the /sim simulator tool (devices/<serial>/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 user backend
topic read devices/# topic read devices/#
topic write devices/+/jobs/ack topic write devices/+/jobs/ack
topic write devices/+/log 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 # testuser is a demo *device*: only the per-device pattern rule above applies

View File

@@ -2,14 +2,17 @@ per_listener_settings true
# Plain MQTT — internal services and clients authenticate with username/password on port 1883 # Plain MQTT — internal services and clients authenticate with username/password on port 1883
listener 1883 0.0.0.0 listener 1883 0.0.0.0
password_file /mosquitto/config/passwd password_file /run/secrets/mosquitto_passwd
acl_file /mosquitto/config/devices.acl acl_file /run/secrets/mosquitto_acl
allow_anonymous false allow_anonymous false
# WebSocket — browser clients, no authentication required # Authenticated MQTT over WebSocket for app clients. Docker binds this listener only to
# host loopback; nginx supplies the public WSS/TLS endpoint at /mqtt on port 443.
listener 9001 0.0.0.0 listener 9001 0.0.0.0
protocol websockets protocol websockets
allow_anonymous true password_file /run/secrets/mosquitto_passwd
acl_file /run/secrets/mosquitto_acl
allow_anonymous false
# TLS MQTT — devices authenticate with client certificates (port 8883) # TLS MQTT — devices authenticate with client certificates (port 8883)
# require_certificate true forces client cert; cert CN becomes the MQTT username. # require_certificate true forces client cert; cert CN becomes the MQTT username.
@@ -19,20 +22,19 @@ allow_anonymous true
# since there's no config/cert hot-reload. # since there's no config/cert hot-reload.
listener 8883 0.0.0.0 listener 8883 0.0.0.0
cafile /mosquitto/certs/ca.crt cafile /mosquitto/certs/ca.crt
certfile /mosquitto/certs/server.crt certfile /mosquitto/certs/public-fullchain.pem
keyfile /mosquitto/certs/server.key keyfile /mosquitto/certs/public-privkey.pem
require_certificate true require_certificate true
use_identity_as_username true use_identity_as_username true
allow_anonymous false allow_anonymous false
acl_file /mosquitto/config/devices.acl acl_file /run/secrets/mosquitto_acl
# TLS MQTT — admin access via username/password, no client cert required (port 8884) # TLS MQTT — app/admin username+password access (port 8884). App usernames are orgIds;
# Connect with CA cert for server verification, then username/password. # devices.acl confines them to ul/{orgId}/app/... . No anonymous listener is exposed.
# listener 8884 0.0.0.0 listener 8884 0.0.0.0
# cafile /mosquitto/certs/ca.crt certfile /mosquitto/certs/public-fullchain.pem
# certfile /mosquitto/certs/server.crt keyfile /mosquitto/certs/public-privkey.pem
# keyfile /mosquitto/certs/server.key require_certificate false
# require_certificate false password_file /run/secrets/mosquitto_passwd
# password_file /mosquitto/config/passwd allow_anonymous false
# allow_anonymous false acl_file /run/secrets/mosquitto_acl
# acl_file /mosquitto/config/devices.acl

View File

@@ -6,18 +6,6 @@ server {
root /var/www/certbot; 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 / { location / {
return 301 https://$host$request_uri; return 301 https://$host$request_uri;
} }
@@ -33,6 +21,8 @@ server {
client_max_body_size 20m; client_max_body_size 20m;
location /mqtt { location /mqtt {
# Authenticated, ACL-confined MQTT-over-WebSocket. Mosquitto is bound to loopback;
# nginx terminates publicly trusted TLS so phones can use standard port 443.
proxy_pass http://127.0.0.1:9001/; proxy_pass http://127.0.0.1:9001/;
proxy_http_version 1.1; proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade; proxy_set_header Upgrade $http_upgrade;