Compare commits
4 Commits
feat/feder
...
a648c8c64d
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a648c8c64d | ||
|
|
9ba79e58c4 | ||
| 6b94d014a8 | |||
| 838c44086c |
@@ -0,0 +1,255 @@
|
|||||||
|
import 'reflect-metadata';
|
||||||
|
import { describe, expect, it, vi } from 'vitest';
|
||||||
|
import type { Db } from '@mosaicstack/db';
|
||||||
|
import type { FederationListResponse } from '@mosaicstack/types';
|
||||||
|
import {
|
||||||
|
FederationClientError,
|
||||||
|
type FederationClientService,
|
||||||
|
} from '../federation-client.service.js';
|
||||||
|
import { type QuerySourceError, QuerySourceService } from '../query-source.service.js';
|
||||||
|
|
||||||
|
interface TestRow {
|
||||||
|
id: string;
|
||||||
|
title: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface PeerRow {
|
||||||
|
id: string;
|
||||||
|
commonName: string;
|
||||||
|
endpointUrl: string | null;
|
||||||
|
clientKeyPem: string | null;
|
||||||
|
state: 'active' | 'pending' | 'suspended' | 'revoked';
|
||||||
|
}
|
||||||
|
|
||||||
|
const LOCAL_ROWS: TestRow[] = [
|
||||||
|
{ id: 'local-1', title: 'Local One' },
|
||||||
|
{ id: 'local-2', title: 'Local Two' },
|
||||||
|
];
|
||||||
|
|
||||||
|
const PEER_A: PeerRow = {
|
||||||
|
id: 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
|
||||||
|
commonName: 'peer-a',
|
||||||
|
endpointUrl: 'https://peer-a.example.com',
|
||||||
|
clientKeyPem: 'sealed-key-a',
|
||||||
|
state: 'active',
|
||||||
|
};
|
||||||
|
|
||||||
|
const PEER_B: PeerRow = {
|
||||||
|
id: 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
|
||||||
|
commonName: 'peer-b',
|
||||||
|
endpointUrl: 'https://peer-b.example.com',
|
||||||
|
clientKeyPem: 'sealed-key-b',
|
||||||
|
state: 'active',
|
||||||
|
};
|
||||||
|
|
||||||
|
const PEER_LOCALHOST: PeerRow = {
|
||||||
|
id: 'cccccccc-cccc-cccc-cccc-cccccccccccc',
|
||||||
|
commonName: 'peer-localhost',
|
||||||
|
endpointUrl: 'https://localhost:3001',
|
||||||
|
clientKeyPem: 'sealed-key-c',
|
||||||
|
state: 'active',
|
||||||
|
};
|
||||||
|
|
||||||
|
function makeDb(activePeers: PeerRow[]): Db {
|
||||||
|
const orderBy = vi.fn().mockResolvedValue(activePeers);
|
||||||
|
const where = vi.fn().mockReturnValue({ orderBy });
|
||||||
|
const from = vi.fn().mockReturnValue({ where });
|
||||||
|
const select = vi.fn().mockReturnValue({ from });
|
||||||
|
|
||||||
|
return {
|
||||||
|
select,
|
||||||
|
insert: vi.fn(),
|
||||||
|
update: vi.fn(),
|
||||||
|
delete: vi.fn(),
|
||||||
|
transaction: vi.fn(),
|
||||||
|
} as unknown as Db;
|
||||||
|
}
|
||||||
|
|
||||||
|
function makeFederationClient(
|
||||||
|
list: (
|
||||||
|
peerId: string,
|
||||||
|
resource: string,
|
||||||
|
request: Record<string, unknown>,
|
||||||
|
) => Promise<FederationListResponse<TestRow>>,
|
||||||
|
): FederationClientService {
|
||||||
|
return {
|
||||||
|
list: list as unknown as FederationClientService['list'],
|
||||||
|
} as FederationClientService;
|
||||||
|
}
|
||||||
|
|
||||||
|
function makeLocalResponse(rows: TestRow[] = LOCAL_ROWS): Promise<FederationListResponse<TestRow>> {
|
||||||
|
return Promise.resolve({ items: rows });
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('QuerySourceService', () => {
|
||||||
|
it('routes source="local" to the local executor and tags rows as local', async () => {
|
||||||
|
const list = vi.fn(async (): Promise<FederationListResponse<TestRow>> => ({ items: [] }));
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_A]), makeFederationClient(list));
|
||||||
|
|
||||||
|
const result = await service.list<TestRow>({
|
||||||
|
source: 'local',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: { cursor: 'ignored-for-local-test' },
|
||||||
|
local: () => makeLocalResponse(),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
items: [
|
||||||
|
{ id: 'local-1', title: 'Local One', _source: 'local' },
|
||||||
|
{ id: 'local-2', title: 'Local Two', _source: 'local' },
|
||||||
|
],
|
||||||
|
});
|
||||||
|
expect(list).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('routes source="federated:<host>" to the matching active peer and tags rows with peer commonName', async () => {
|
||||||
|
const list = vi.fn(
|
||||||
|
async (): Promise<FederationListResponse<TestRow>> => ({
|
||||||
|
items: [{ id: 'remote-1', title: 'Remote One' }],
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_A, PEER_B]), makeFederationClient(list));
|
||||||
|
|
||||||
|
const result = await service.list<TestRow>({
|
||||||
|
source: 'federated:peer-b.example.com',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: { status: 'open' },
|
||||||
|
local: () => makeLocalResponse(),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
items: [{ id: 'remote-1', title: 'Remote One', _source: 'peer-b' }],
|
||||||
|
});
|
||||||
|
expect(list).toHaveBeenCalledWith(PEER_B.id, 'tasks', { status: 'open' });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('matches federated hosts by endpoint host including non-default port', async () => {
|
||||||
|
const list = vi.fn(
|
||||||
|
async (): Promise<FederationListResponse<TestRow>> => ({
|
||||||
|
items: [{ id: 'remote-port', title: 'Remote Port' }],
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_LOCALHOST]), makeFederationClient(list));
|
||||||
|
|
||||||
|
const result = await service.list<TestRow>({
|
||||||
|
source: 'federated:localhost:3001',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: {},
|
||||||
|
local: () => makeLocalResponse(),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
items: [{ id: 'remote-port', title: 'Remote Port', _source: 'peer-localhost' }],
|
||||||
|
});
|
||||||
|
expect(list).toHaveBeenCalledWith(PEER_LOCALHOST.id, 'tasks', {});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('fans out source="all" to local plus every active outbound peer in parallel and merges tagged rows', async () => {
|
||||||
|
const callOrder: string[] = [];
|
||||||
|
const list = vi.fn(async (peerId: string): Promise<FederationListResponse<TestRow>> => {
|
||||||
|
callOrder.push(`remote-start:${peerId}`);
|
||||||
|
await Promise.resolve();
|
||||||
|
return {
|
||||||
|
items: [{ id: `remote-${peerId.slice(0, 1)}`, title: `Remote ${peerId.slice(0, 1)}` }],
|
||||||
|
};
|
||||||
|
});
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_A, PEER_B]), makeFederationClient(list));
|
||||||
|
|
||||||
|
const result = await service.list<TestRow>({
|
||||||
|
source: 'all',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: { limit: 25 },
|
||||||
|
local: async () => {
|
||||||
|
callOrder.push('local-start');
|
||||||
|
await Promise.resolve();
|
||||||
|
return { items: [{ id: 'local-1', title: 'Local One' }] };
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
items: [
|
||||||
|
{ id: 'local-1', title: 'Local One', _source: 'local' },
|
||||||
|
{ id: 'remote-a', title: 'Remote a', _source: 'peer-a' },
|
||||||
|
{ id: 'remote-b', title: 'Remote b', _source: 'peer-b' },
|
||||||
|
],
|
||||||
|
});
|
||||||
|
expect(list).toHaveBeenCalledTimes(2);
|
||||||
|
expect(callOrder).toEqual([
|
||||||
|
'local-start',
|
||||||
|
`remote-start:${PEER_A.id}`,
|
||||||
|
`remote-start:${PEER_B.id}`,
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('marks source="all" as partial and truncated when any subquery returns a cursor', async () => {
|
||||||
|
const list = vi.fn(
|
||||||
|
async (): Promise<FederationListResponse<TestRow>> => ({
|
||||||
|
items: [{ id: 'remote-a', title: 'Remote A' }],
|
||||||
|
nextCursor: 'remote-next',
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_A]), makeFederationClient(list));
|
||||||
|
|
||||||
|
const result = await service.list<TestRow>({
|
||||||
|
source: 'all',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: {},
|
||||||
|
local: () => makeLocalResponse([{ id: 'local-1', title: 'Local One' }]),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
items: [
|
||||||
|
{ id: 'local-1', title: 'Local One', _source: 'local' },
|
||||||
|
{ id: 'remote-a', title: 'Remote A', _source: 'peer-a' },
|
||||||
|
],
|
||||||
|
_partial: true,
|
||||||
|
_truncated: true,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('returns _partial=true for source="all" when one peer fails without dropping successful sources', async () => {
|
||||||
|
const list = vi.fn(async (peerId: string): Promise<FederationListResponse<TestRow>> => {
|
||||||
|
if (peerId === PEER_B.id) {
|
||||||
|
throw new FederationClientError({
|
||||||
|
code: 'NETWORK',
|
||||||
|
message: 'peer unavailable',
|
||||||
|
peerId,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
return { items: [{ id: 'remote-a', title: 'Remote A' }] };
|
||||||
|
});
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_A, PEER_B]), makeFederationClient(list));
|
||||||
|
|
||||||
|
const result = await service.list<TestRow>({
|
||||||
|
source: 'all',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: {},
|
||||||
|
local: () => makeLocalResponse([{ id: 'local-1', title: 'Local One' }]),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
items: [
|
||||||
|
{ id: 'local-1', title: 'Local One', _source: 'local' },
|
||||||
|
{ id: 'remote-a', title: 'Remote A', _source: 'peer-a' },
|
||||||
|
],
|
||||||
|
_partial: true,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('throws QuerySourceError when a federated host does not match an active outbound peer', async () => {
|
||||||
|
const list = vi.fn(async (): Promise<FederationListResponse<TestRow>> => ({ items: [] }));
|
||||||
|
const service = new QuerySourceService(makeDb([PEER_A]), makeFederationClient(list));
|
||||||
|
|
||||||
|
await expect(
|
||||||
|
service.list<TestRow>({
|
||||||
|
source: 'federated:missing.example.com',
|
||||||
|
resource: 'tasks',
|
||||||
|
request: {},
|
||||||
|
local: () => makeLocalResponse(),
|
||||||
|
}),
|
||||||
|
).rejects.toMatchObject({
|
||||||
|
name: 'QuerySourceError',
|
||||||
|
code: 'PEER_NOT_FOUND',
|
||||||
|
} satisfies Partial<QuerySourceError>);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -11,3 +11,13 @@ export {
|
|||||||
type FederationClientErrorCode,
|
type FederationClientErrorCode,
|
||||||
type FederationClientErrorOptions,
|
type FederationClientErrorOptions,
|
||||||
} from './federation-client.service.js';
|
} from './federation-client.service.js';
|
||||||
|
export {
|
||||||
|
QuerySourceService,
|
||||||
|
QuerySourceError,
|
||||||
|
type QuerySource,
|
||||||
|
type QuerySourceErrorCode,
|
||||||
|
type QuerySourceErrorOptions,
|
||||||
|
type QuerySourceListOptions,
|
||||||
|
type QuerySourceListResponse,
|
||||||
|
type LocalListExecutor,
|
||||||
|
} from './query-source.service.js';
|
||||||
|
|||||||
261
apps/gateway/src/federation/client/query-source.service.ts
Normal file
261
apps/gateway/src/federation/client/query-source.service.ts
Normal file
@@ -0,0 +1,261 @@
|
|||||||
|
/**
|
||||||
|
* QuerySourceService — gateway query source router (FED-M3-09).
|
||||||
|
*
|
||||||
|
* Accepts the federation query-layer `source` selector and routes list-style
|
||||||
|
* reads to local storage, one federated peer, or all active outbound peers.
|
||||||
|
*
|
||||||
|
* `source: "all"` is intentionally tolerant of per-peer failures: local data
|
||||||
|
* and successful peer responses are returned, and the envelope is marked
|
||||||
|
* `_partial: true`. Local failures still reject because there is no safe local
|
||||||
|
* fallback and the gateway's own storage is expected to be authoritative.
|
||||||
|
*/
|
||||||
|
|
||||||
|
import { Inject, Injectable, Logger } from '@nestjs/common';
|
||||||
|
import { and, eq, federationPeers, isNotNull, type Db } from '@mosaicstack/db';
|
||||||
|
import {
|
||||||
|
SOURCE_LOCAL,
|
||||||
|
tagWithSource,
|
||||||
|
type FederationListResponse,
|
||||||
|
type SourceTag,
|
||||||
|
} from '@mosaicstack/types';
|
||||||
|
import { DB } from '../../database/database.module.js';
|
||||||
|
import { FederationClientService } from './federation-client.service.js';
|
||||||
|
|
||||||
|
export type QuerySource = 'local' | 'all' | `federated:${string}`;
|
||||||
|
|
||||||
|
export type QuerySourceErrorCode = 'INVALID_SOURCE' | 'PEER_NOT_FOUND';
|
||||||
|
|
||||||
|
export interface QuerySourceErrorOptions {
|
||||||
|
code: QuerySourceErrorCode;
|
||||||
|
message: string;
|
||||||
|
source: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export class QuerySourceError extends Error {
|
||||||
|
readonly code: QuerySourceErrorCode;
|
||||||
|
readonly source: string;
|
||||||
|
|
||||||
|
constructor(opts: QuerySourceErrorOptions) {
|
||||||
|
super(opts.message);
|
||||||
|
this.name = 'QuerySourceError';
|
||||||
|
this.code = opts.code;
|
||||||
|
this.source = opts.source;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export type LocalListExecutor<T extends object> = () => Promise<FederationListResponse<T> | T[]>;
|
||||||
|
|
||||||
|
export interface QuerySourceListOptions<T extends object> {
|
||||||
|
source: QuerySource;
|
||||||
|
resource: string;
|
||||||
|
request?: Record<string, unknown>;
|
||||||
|
local: LocalListExecutor<T>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export type QuerySourceListResponse<T extends object> = FederationListResponse<T & SourceTag>;
|
||||||
|
|
||||||
|
interface OutboundPeer {
|
||||||
|
id: string;
|
||||||
|
commonName: string;
|
||||||
|
endpointUrl: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface TaggedList<T extends object> {
|
||||||
|
items: Array<T & SourceTag>;
|
||||||
|
partial: boolean;
|
||||||
|
truncated: boolean;
|
||||||
|
nextCursor?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class QuerySourceService {
|
||||||
|
private readonly logger = new Logger(QuerySourceService.name);
|
||||||
|
|
||||||
|
constructor(
|
||||||
|
@Inject(DB) private readonly db: Db,
|
||||||
|
@Inject(FederationClientService) private readonly federationClient: FederationClientService,
|
||||||
|
) {}
|
||||||
|
|
||||||
|
async list<T extends object>(
|
||||||
|
options: QuerySourceListOptions<T>,
|
||||||
|
): Promise<QuerySourceListResponse<T>> {
|
||||||
|
const request = options.request ?? {};
|
||||||
|
|
||||||
|
if (options.source === 'local') {
|
||||||
|
const local = await this.runLocal(options.local);
|
||||||
|
return this.toResponse(this.tagList(local, SOURCE_LOCAL));
|
||||||
|
}
|
||||||
|
|
||||||
|
if (options.source === 'all') {
|
||||||
|
return this.listAll(options.resource, request, options.local);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (options.source.startsWith('federated:')) {
|
||||||
|
const host = options.source.slice('federated:'.length).trim();
|
||||||
|
if (!host) {
|
||||||
|
throw new QuerySourceError({
|
||||||
|
code: 'INVALID_SOURCE',
|
||||||
|
message: 'Federated source must include a host after federated:',
|
||||||
|
source: options.source,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
const peer = await this.findPeerByHost(host, options.source);
|
||||||
|
const remote = await this.federationClient.list<T>(peer.id, options.resource, request);
|
||||||
|
return this.toResponse(this.tagList(remote, peer.commonName));
|
||||||
|
}
|
||||||
|
|
||||||
|
throw new QuerySourceError({
|
||||||
|
code: 'INVALID_SOURCE',
|
||||||
|
message: `Unsupported query source: ${options.source}`,
|
||||||
|
source: options.source,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
private async listAll<T extends object>(
|
||||||
|
resource: string,
|
||||||
|
request: Record<string, unknown>,
|
||||||
|
local: LocalListExecutor<T>,
|
||||||
|
): Promise<QuerySourceListResponse<T>> {
|
||||||
|
const peers = await this.listActiveOutboundPeers();
|
||||||
|
|
||||||
|
const localPromise = this.runLocal(local).then((response) =>
|
||||||
|
this.tagList(response, SOURCE_LOCAL),
|
||||||
|
);
|
||||||
|
const remotePromises = peers.map(async (peer: OutboundPeer): Promise<TaggedList<T> | null> => {
|
||||||
|
try {
|
||||||
|
const response = await this.federationClient.list<T>(peer.id, resource, request);
|
||||||
|
return this.tagList(response, peer.commonName);
|
||||||
|
} catch (error: unknown) {
|
||||||
|
this.logger.warn(
|
||||||
|
`Federated query to peer ${peer.commonName} (${peer.id}) failed; returning partial all-source response: ${
|
||||||
|
error instanceof Error ? error.message : String(error)
|
||||||
|
}`,
|
||||||
|
);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
const [localResult, ...remoteResults] = await Promise.all([localPromise, ...remotePromises]);
|
||||||
|
const successfulRemoteResults = remoteResults.filter(
|
||||||
|
(result: TaggedList<T> | null): result is TaggedList<T> => result !== null,
|
||||||
|
);
|
||||||
|
const allResults = [localResult, ...successfulRemoteResults];
|
||||||
|
const peerFailure = successfulRemoteResults.length !== peers.length;
|
||||||
|
|
||||||
|
return this.mergeTaggedLists(allResults, peerFailure);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async runLocal<T extends object>(
|
||||||
|
local: LocalListExecutor<T>,
|
||||||
|
): Promise<FederationListResponse<T>> {
|
||||||
|
const response = await local();
|
||||||
|
if (Array.isArray(response)) {
|
||||||
|
return { items: response };
|
||||||
|
}
|
||||||
|
return response;
|
||||||
|
}
|
||||||
|
|
||||||
|
private tagList<T extends object>(
|
||||||
|
response: FederationListResponse<T>,
|
||||||
|
source: string,
|
||||||
|
): TaggedList<T> {
|
||||||
|
return {
|
||||||
|
items: tagWithSource(response.items, source),
|
||||||
|
partial: response._partial === true,
|
||||||
|
truncated: response._truncated === true || response.nextCursor !== undefined,
|
||||||
|
nextCursor: response.nextCursor,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
private mergeTaggedLists<T extends object>(
|
||||||
|
lists: Array<TaggedList<T>>,
|
||||||
|
peerFailure: boolean,
|
||||||
|
): QuerySourceListResponse<T> {
|
||||||
|
const items = lists.flatMap((list: TaggedList<T>) => list.items);
|
||||||
|
const partial =
|
||||||
|
peerFailure ||
|
||||||
|
lists.some((list: TaggedList<T>) => list.partial || list.nextCursor !== undefined);
|
||||||
|
const truncated = lists.some((list: TaggedList<T>) => list.truncated);
|
||||||
|
|
||||||
|
const response: QuerySourceListResponse<T> = { items };
|
||||||
|
if (partial) {
|
||||||
|
response._partial = true;
|
||||||
|
}
|
||||||
|
if (truncated) {
|
||||||
|
response._truncated = true;
|
||||||
|
}
|
||||||
|
return response;
|
||||||
|
}
|
||||||
|
|
||||||
|
private toResponse<T extends object>(tagged: TaggedList<T>): QuerySourceListResponse<T> {
|
||||||
|
const response: QuerySourceListResponse<T> = {
|
||||||
|
items: tagged.items,
|
||||||
|
};
|
||||||
|
if (tagged.nextCursor !== undefined) {
|
||||||
|
response.nextCursor = tagged.nextCursor;
|
||||||
|
}
|
||||||
|
if (tagged.partial) {
|
||||||
|
response._partial = true;
|
||||||
|
}
|
||||||
|
if (tagged.truncated) {
|
||||||
|
response._truncated = true;
|
||||||
|
}
|
||||||
|
return response;
|
||||||
|
}
|
||||||
|
|
||||||
|
private async findPeerByHost(sourceHost: string, source: string): Promise<OutboundPeer> {
|
||||||
|
const host = normalizeHost(sourceHost);
|
||||||
|
const peers = await this.listActiveOutboundPeers();
|
||||||
|
const peer = peers.find((candidate: OutboundPeer) => {
|
||||||
|
const commonName = normalizeHost(candidate.commonName);
|
||||||
|
const endpointHosts = endpointHostKeys(candidate.endpointUrl).map((endpointHost: string) =>
|
||||||
|
normalizeHost(endpointHost),
|
||||||
|
);
|
||||||
|
return commonName === host || endpointHosts.includes(host);
|
||||||
|
});
|
||||||
|
|
||||||
|
if (!peer) {
|
||||||
|
throw new QuerySourceError({
|
||||||
|
code: 'PEER_NOT_FOUND',
|
||||||
|
message: `No active outbound federation peer matches source ${source}`,
|
||||||
|
source,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
return peer;
|
||||||
|
}
|
||||||
|
|
||||||
|
private async listActiveOutboundPeers(): Promise<OutboundPeer[]> {
|
||||||
|
const rows = await this.db
|
||||||
|
.select({
|
||||||
|
id: federationPeers.id,
|
||||||
|
commonName: federationPeers.commonName,
|
||||||
|
endpointUrl: federationPeers.endpointUrl,
|
||||||
|
})
|
||||||
|
.from(federationPeers)
|
||||||
|
.where(
|
||||||
|
and(
|
||||||
|
eq(federationPeers.state, 'active'),
|
||||||
|
isNotNull(federationPeers.endpointUrl),
|
||||||
|
isNotNull(federationPeers.clientKeyPem),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.orderBy(federationPeers.commonName);
|
||||||
|
|
||||||
|
return rows.filter((row): row is OutboundPeer => typeof row.endpointUrl === 'string');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeHost(host: string): string {
|
||||||
|
return host.trim().toLowerCase();
|
||||||
|
}
|
||||||
|
|
||||||
|
function endpointHostKeys(endpointUrl: string): string[] {
|
||||||
|
try {
|
||||||
|
const url = new URL(endpointUrl);
|
||||||
|
return Array.from(new Set([url.host, url.hostname].filter((host: string) => host.length > 0)));
|
||||||
|
} catch {
|
||||||
|
return [];
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -6,8 +6,8 @@ import { EnrollmentService } from './enrollment.service.js';
|
|||||||
import { FederationController } from './federation.controller.js';
|
import { FederationController } from './federation.controller.js';
|
||||||
import { CapabilitiesController } from './server/verbs/capabilities.controller.js';
|
import { CapabilitiesController } from './server/verbs/capabilities.controller.js';
|
||||||
import { GrantsService } from './grants.service.js';
|
import { GrantsService } from './grants.service.js';
|
||||||
import { FederationClientService } from './client/index.js';
|
import { FederationClientService, QuerySourceService } from './client/index.js';
|
||||||
import { FederationAuthGuard } from './server/index.js';
|
import { FederationAuthGuard, FederationScopeService } from './server/index.js';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
controllers: [EnrollmentController, FederationController, CapabilitiesController],
|
controllers: [EnrollmentController, FederationController, CapabilitiesController],
|
||||||
@@ -17,14 +17,18 @@ import { FederationAuthGuard } from './server/index.js';
|
|||||||
EnrollmentService,
|
EnrollmentService,
|
||||||
GrantsService,
|
GrantsService,
|
||||||
FederationClientService,
|
FederationClientService,
|
||||||
|
QuerySourceService,
|
||||||
FederationAuthGuard,
|
FederationAuthGuard,
|
||||||
|
FederationScopeService,
|
||||||
],
|
],
|
||||||
exports: [
|
exports: [
|
||||||
CaService,
|
CaService,
|
||||||
EnrollmentService,
|
EnrollmentService,
|
||||||
GrantsService,
|
GrantsService,
|
||||||
FederationClientService,
|
FederationClientService,
|
||||||
|
QuerySourceService,
|
||||||
FederationAuthGuard,
|
FederationAuthGuard,
|
||||||
|
FederationScopeService,
|
||||||
],
|
],
|
||||||
})
|
})
|
||||||
export class FederationModule {}
|
export class FederationModule {}
|
||||||
|
|||||||
@@ -0,0 +1,324 @@
|
|||||||
|
/**
|
||||||
|
* Unit tests for FederationScopeService (FED-M3-04).
|
||||||
|
*
|
||||||
|
* Coverage:
|
||||||
|
* - resource allowlist deny
|
||||||
|
* - excluded resource deny
|
||||||
|
* - invalid scope deny
|
||||||
|
* - invalid requested limit deny
|
||||||
|
* - native RBAC deny as subjectUserId
|
||||||
|
* - scope/native filter intersection for personal and team rows
|
||||||
|
* - native RBAC personal deny wins over scope include_personal allow/default
|
||||||
|
* - max_rows_per_query cap
|
||||||
|
*/
|
||||||
|
|
||||||
|
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||||
|
import { FederationScopeService, type FederationNativeRbacEvaluator } from '../scope.service.js';
|
||||||
|
import type { FederationContext } from '../federation-context.js';
|
||||||
|
|
||||||
|
const GRANT_ID = 'grant-1';
|
||||||
|
const PEER_ID = 'peer-1';
|
||||||
|
const SUBJECT_USER_ID = 'user-1';
|
||||||
|
|
||||||
|
function makeContext(scope: Record<string, unknown>): FederationContext {
|
||||||
|
return {
|
||||||
|
grantId: GRANT_ID,
|
||||||
|
peerId: PEER_ID,
|
||||||
|
subjectUserId: SUBJECT_USER_ID,
|
||||||
|
scope,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function makeNativeRbac(
|
||||||
|
result: Awaited<ReturnType<FederationNativeRbacEvaluator['evaluateReadAccess']>>,
|
||||||
|
): FederationNativeRbacEvaluator {
|
||||||
|
return {
|
||||||
|
evaluateReadAccess: vi.fn().mockResolvedValue(result),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('FederationScopeService', () => {
|
||||||
|
let service: FederationScopeService;
|
||||||
|
|
||||||
|
beforeEach(() => {
|
||||||
|
service = new FederationScopeService();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('allows a granted resource and returns a capped query filter', async () => {
|
||||||
|
const nativeRbac = makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: ['team-1', 'team-2'] },
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({
|
||||||
|
resources: ['tasks'],
|
||||||
|
filters: { tasks: { include_teams: ['team-1', 'team-3'], include_personal: true } },
|
||||||
|
max_rows_per_query: 50,
|
||||||
|
}),
|
||||||
|
resource: 'tasks',
|
||||||
|
requestedLimit: 500,
|
||||||
|
nativeRbac,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
allowed: true,
|
||||||
|
filter: {
|
||||||
|
resource: 'tasks',
|
||||||
|
subjectUserId: SUBJECT_USER_ID,
|
||||||
|
includePersonal: true,
|
||||||
|
teamIds: ['team-1'],
|
||||||
|
limit: 50,
|
||||||
|
maxRowsPerQuery: 50,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(nativeRbac.evaluateReadAccess).toHaveBeenCalledWith({
|
||||||
|
grantId: GRANT_ID,
|
||||||
|
peerId: PEER_ID,
|
||||||
|
subjectUserId: SUBJECT_USER_ID,
|
||||||
|
resource: 'tasks',
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('defaults absent resource filters to native RBAC personal and team visibility', async () => {
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({ resources: ['notes'], max_rows_per_query: 100 }),
|
||||||
|
resource: 'notes',
|
||||||
|
nativeRbac: makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: ['team-1', 'team-2'] },
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: true,
|
||||||
|
filter: {
|
||||||
|
includePersonal: true,
|
||||||
|
teamIds: ['team-1', 'team-2'],
|
||||||
|
limit: 100,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('honors include_personal false even when native RBAC allows personal rows', async () => {
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({
|
||||||
|
resources: ['memory'],
|
||||||
|
filters: { memory: { include_personal: false } },
|
||||||
|
max_rows_per_query: 25,
|
||||||
|
}),
|
||||||
|
resource: 'memory',
|
||||||
|
nativeRbac: makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: [] },
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: true,
|
||||||
|
filter: {
|
||||||
|
includePersonal: false,
|
||||||
|
teamIds: [],
|
||||||
|
},
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not leak personal rows when scope allows personal but native RBAC denies personal', async () => {
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({
|
||||||
|
resources: ['tasks'],
|
||||||
|
filters: { tasks: { include_personal: true } },
|
||||||
|
max_rows_per_query: 25,
|
||||||
|
}),
|
||||||
|
resource: 'tasks',
|
||||||
|
nativeRbac: makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: false, teamIds: ['team-1'] },
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: true,
|
||||||
|
filter: {
|
||||||
|
includePersonal: false,
|
||||||
|
teamIds: ['team-1'],
|
||||||
|
},
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not widen native RBAC when scope includes teams the user cannot access', async () => {
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({
|
||||||
|
resources: ['tasks'],
|
||||||
|
filters: { tasks: { include_teams: ['team-2'], include_personal: false } },
|
||||||
|
max_rows_per_query: 25,
|
||||||
|
}),
|
||||||
|
resource: 'tasks',
|
||||||
|
nativeRbac: makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: ['team-1'] },
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: true,
|
||||||
|
filter: {
|
||||||
|
includePersonal: false,
|
||||||
|
teamIds: [],
|
||||||
|
},
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('denies invalid grant scope before RBAC evaluation', async () => {
|
||||||
|
const nativeRbac = makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: [] },
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({ resources: [], max_rows_per_query: 100 }),
|
||||||
|
resource: 'tasks',
|
||||||
|
nativeRbac,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: 'invalid_scope',
|
||||||
|
stage: 'scope_parse',
|
||||||
|
statusCode: 400,
|
||||||
|
grantId: GRANT_ID,
|
||||||
|
subjectUserId: SUBJECT_USER_ID,
|
||||||
|
resource: 'tasks',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(nativeRbac.evaluateReadAccess).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('denies unsupported resource names before RBAC evaluation', async () => {
|
||||||
|
const nativeRbac = makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: [] },
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({ resources: ['tasks'], max_rows_per_query: 100 }),
|
||||||
|
resource: 'unknown_resource',
|
||||||
|
nativeRbac,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: 'invalid_resource',
|
||||||
|
stage: 'resource_allowlist',
|
||||||
|
statusCode: 403,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(nativeRbac.evaluateReadAccess).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('denies resources explicitly present in excluded_resources before allowlist miss', async () => {
|
||||||
|
const nativeRbac = makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: [] },
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({
|
||||||
|
resources: ['tasks'],
|
||||||
|
excluded_resources: ['credentials'],
|
||||||
|
max_rows_per_query: 100,
|
||||||
|
}),
|
||||||
|
resource: 'credentials',
|
||||||
|
nativeRbac,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: 'resource_excluded',
|
||||||
|
stage: 'resource_exclusion',
|
||||||
|
statusCode: 403,
|
||||||
|
resource: 'credentials',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(nativeRbac.evaluateReadAccess).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('denies supported resources that are not granted by scope', async () => {
|
||||||
|
const nativeRbac = makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: [] },
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({ resources: ['tasks'], max_rows_per_query: 100 }),
|
||||||
|
resource: 'notes',
|
||||||
|
nativeRbac,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: 'resource_not_granted',
|
||||||
|
stage: 'resource_allowlist',
|
||||||
|
statusCode: 403,
|
||||||
|
resource: 'notes',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(nativeRbac.evaluateReadAccess).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('denies invalid requested row limits before RBAC evaluation', async () => {
|
||||||
|
const nativeRbac = makeNativeRbac({
|
||||||
|
allowed: true,
|
||||||
|
access: { includePersonal: true, teamIds: [] },
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({ resources: ['tasks'], max_rows_per_query: 100 }),
|
||||||
|
resource: 'tasks',
|
||||||
|
requestedLimit: 0,
|
||||||
|
nativeRbac,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: 'invalid_limit',
|
||||||
|
stage: 'row_cap',
|
||||||
|
statusCode: 400,
|
||||||
|
details: { requestedLimit: 0 },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(nativeRbac.evaluateReadAccess).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('denies when native RBAC rejects subjectUserId access to the resource', async () => {
|
||||||
|
const result = await service.evaluateAccess({
|
||||||
|
context: makeContext({ resources: ['tasks'], max_rows_per_query: 100 }),
|
||||||
|
resource: 'tasks',
|
||||||
|
nativeRbac: makeNativeRbac({
|
||||||
|
allowed: false,
|
||||||
|
reason: 'read:tasks denied',
|
||||||
|
details: { permission: 'tasks:read' },
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result).toEqual({
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: 'native_rbac_denied',
|
||||||
|
stage: 'native_rbac',
|
||||||
|
statusCode: 403,
|
||||||
|
message: 'read:tasks denied',
|
||||||
|
grantId: GRANT_ID,
|
||||||
|
peerId: PEER_ID,
|
||||||
|
subjectUserId: SUBJECT_USER_ID,
|
||||||
|
resource: 'tasks',
|
||||||
|
details: { permission: 'tasks:read' },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -10,4 +10,22 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
export { FederationAuthGuard } from './federation-auth.guard.js';
|
export { FederationAuthGuard } from './federation-auth.guard.js';
|
||||||
|
export { FederationScopeService } from './scope.service.js';
|
||||||
export type { FederationContext } from './federation-context.js';
|
export type { FederationContext } from './federation-context.js';
|
||||||
|
export type {
|
||||||
|
FederationNativeRbacAccess,
|
||||||
|
FederationNativeRbacAllowedResult,
|
||||||
|
FederationNativeRbacDeniedResult,
|
||||||
|
FederationNativeRbacEvaluator,
|
||||||
|
FederationNativeRbacRequest,
|
||||||
|
FederationNativeRbacResult,
|
||||||
|
FederationScopeAllowedResult,
|
||||||
|
FederationScopeDeniedResult,
|
||||||
|
FederationScopeDenyCode,
|
||||||
|
FederationScopeDenyDetails,
|
||||||
|
FederationScopeDenyReason,
|
||||||
|
FederationScopeDenyStage,
|
||||||
|
FederationScopeEvaluationInput,
|
||||||
|
FederationScopeEvaluationResult,
|
||||||
|
FederationScopeQueryFilter,
|
||||||
|
} from './scope.service.js';
|
||||||
|
|||||||
272
apps/gateway/src/federation/server/scope.service.ts
Normal file
272
apps/gateway/src/federation/server/scope.service.ts
Normal file
@@ -0,0 +1,272 @@
|
|||||||
|
/**
|
||||||
|
* FederationScopeService — M3 server-side scope enforcement pipeline.
|
||||||
|
*
|
||||||
|
* Pure trust-boundary service: it validates the grant scope, asks an injected
|
||||||
|
* native RBAC evaluator what the subject user can read locally, intersects that
|
||||||
|
* answer with the federation scope filters, and returns a query filter for the
|
||||||
|
* verb controllers. The service performs no DB calls directly.
|
||||||
|
*/
|
||||||
|
|
||||||
|
import { Injectable } from '@nestjs/common';
|
||||||
|
import {
|
||||||
|
FEDERATION_RESOURCE_VALUES,
|
||||||
|
type FederationResource,
|
||||||
|
FederationScopeError,
|
||||||
|
parseFederationScope,
|
||||||
|
} from '../scope-schema.js';
|
||||||
|
import type { FederationContext } from './federation-context.js';
|
||||||
|
|
||||||
|
const federationResourceSet: ReadonlySet<string> = new Set<string>(FEDERATION_RESOURCE_VALUES);
|
||||||
|
|
||||||
|
export type FederationScopeDenyStage =
|
||||||
|
| 'scope_parse'
|
||||||
|
| 'resource_allowlist'
|
||||||
|
| 'resource_exclusion'
|
||||||
|
| 'native_rbac'
|
||||||
|
| 'row_cap';
|
||||||
|
|
||||||
|
export type FederationScopeDenyCode =
|
||||||
|
| 'invalid_scope'
|
||||||
|
| 'invalid_resource'
|
||||||
|
| 'resource_not_granted'
|
||||||
|
| 'resource_excluded'
|
||||||
|
| 'native_rbac_denied'
|
||||||
|
| 'invalid_limit';
|
||||||
|
|
||||||
|
export type FederationScopeDenyStatus = 400 | 403;
|
||||||
|
|
||||||
|
export interface FederationScopeDenyDetails {
|
||||||
|
readonly [key: string]: string | number | boolean | readonly string[];
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationScopeDenyReason {
|
||||||
|
readonly code: FederationScopeDenyCode;
|
||||||
|
readonly stage: FederationScopeDenyStage;
|
||||||
|
readonly statusCode: FederationScopeDenyStatus;
|
||||||
|
readonly message: string;
|
||||||
|
readonly grantId: string;
|
||||||
|
readonly peerId: string;
|
||||||
|
readonly subjectUserId: string;
|
||||||
|
readonly resource: string;
|
||||||
|
readonly details?: FederationScopeDenyDetails;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationNativeRbacRequest {
|
||||||
|
readonly grantId: string;
|
||||||
|
readonly peerId: string;
|
||||||
|
readonly subjectUserId: string;
|
||||||
|
readonly resource: FederationResource;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationNativeRbacAccess {
|
||||||
|
/** Whether this user may read personal rows for this resource. */
|
||||||
|
readonly includePersonal: boolean;
|
||||||
|
|
||||||
|
/** Team IDs this user may read for this resource under native RBAC. */
|
||||||
|
readonly teamIds: readonly string[];
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationNativeRbacAllowedResult {
|
||||||
|
readonly allowed: true;
|
||||||
|
readonly access: FederationNativeRbacAccess;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationNativeRbacDeniedResult {
|
||||||
|
readonly allowed: false;
|
||||||
|
readonly reason?: string;
|
||||||
|
readonly details?: FederationScopeDenyDetails;
|
||||||
|
}
|
||||||
|
|
||||||
|
export type FederationNativeRbacResult =
|
||||||
|
| FederationNativeRbacAllowedResult
|
||||||
|
| FederationNativeRbacDeniedResult;
|
||||||
|
|
||||||
|
export interface FederationNativeRbacEvaluator {
|
||||||
|
evaluateReadAccess(request: FederationNativeRbacRequest): Promise<FederationNativeRbacResult>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationScopeEvaluationInput {
|
||||||
|
readonly context: FederationContext;
|
||||||
|
readonly resource: string;
|
||||||
|
readonly requestedLimit?: number;
|
||||||
|
readonly nativeRbac: FederationNativeRbacEvaluator;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationScopeQueryFilter {
|
||||||
|
readonly resource: FederationResource;
|
||||||
|
readonly subjectUserId: string;
|
||||||
|
readonly includePersonal: boolean;
|
||||||
|
readonly teamIds: readonly string[];
|
||||||
|
readonly limit: number;
|
||||||
|
readonly maxRowsPerQuery: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationScopeAllowedResult {
|
||||||
|
readonly allowed: true;
|
||||||
|
readonly filter: FederationScopeQueryFilter;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface FederationScopeDeniedResult {
|
||||||
|
readonly allowed: false;
|
||||||
|
readonly deny: FederationScopeDenyReason;
|
||||||
|
}
|
||||||
|
|
||||||
|
export type FederationScopeEvaluationResult =
|
||||||
|
| FederationScopeAllowedResult
|
||||||
|
| FederationScopeDeniedResult;
|
||||||
|
|
||||||
|
function isFederationResource(resource: string): resource is FederationResource {
|
||||||
|
return federationResourceSet.has(resource);
|
||||||
|
}
|
||||||
|
|
||||||
|
function uniqueStrings(values: readonly string[]): readonly string[] {
|
||||||
|
return Array.from(new Set<string>(values));
|
||||||
|
}
|
||||||
|
|
||||||
|
function intersectTeamIds(
|
||||||
|
nativeTeamIds: readonly string[],
|
||||||
|
scopedTeamIds: readonly string[] | undefined,
|
||||||
|
): readonly string[] {
|
||||||
|
const uniqueNativeTeamIds = uniqueStrings(nativeTeamIds);
|
||||||
|
|
||||||
|
if (scopedTeamIds === undefined) {
|
||||||
|
return uniqueNativeTeamIds;
|
||||||
|
}
|
||||||
|
|
||||||
|
const nativeSet = new Set<string>(uniqueNativeTeamIds);
|
||||||
|
return uniqueStrings(scopedTeamIds).filter((teamId: string): boolean => nativeSet.has(teamId));
|
||||||
|
}
|
||||||
|
|
||||||
|
function makeDenyReason(params: {
|
||||||
|
readonly code: FederationScopeDenyCode;
|
||||||
|
readonly stage: FederationScopeDenyStage;
|
||||||
|
readonly statusCode?: FederationScopeDenyStatus;
|
||||||
|
readonly message: string;
|
||||||
|
readonly context: FederationContext;
|
||||||
|
readonly resource: string;
|
||||||
|
readonly details?: FederationScopeDenyDetails;
|
||||||
|
}): FederationScopeDeniedResult {
|
||||||
|
return {
|
||||||
|
allowed: false,
|
||||||
|
deny: {
|
||||||
|
code: params.code,
|
||||||
|
stage: params.stage,
|
||||||
|
statusCode: params.statusCode ?? 403,
|
||||||
|
message: params.message,
|
||||||
|
grantId: params.context.grantId,
|
||||||
|
peerId: params.context.peerId,
|
||||||
|
subjectUserId: params.context.subjectUserId,
|
||||||
|
resource: params.resource,
|
||||||
|
...(params.details !== undefined ? { details: params.details } : {}),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class FederationScopeService {
|
||||||
|
async evaluateAccess(
|
||||||
|
input: FederationScopeEvaluationInput,
|
||||||
|
): Promise<FederationScopeEvaluationResult> {
|
||||||
|
const { context, resource, requestedLimit, nativeRbac } = input;
|
||||||
|
|
||||||
|
let scope: ReturnType<typeof parseFederationScope>;
|
||||||
|
try {
|
||||||
|
scope = parseFederationScope(context.scope);
|
||||||
|
} catch (error: unknown) {
|
||||||
|
const message =
|
||||||
|
error instanceof FederationScopeError
|
||||||
|
? 'Federation grant scope is invalid'
|
||||||
|
: 'Federation grant scope could not be parsed';
|
||||||
|
const details = error instanceof Error ? { reason: error.message } : undefined;
|
||||||
|
return makeDenyReason({
|
||||||
|
code: 'invalid_scope',
|
||||||
|
stage: 'scope_parse',
|
||||||
|
statusCode: 400,
|
||||||
|
message,
|
||||||
|
context,
|
||||||
|
resource,
|
||||||
|
...(details !== undefined ? { details } : {}),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!isFederationResource(resource)) {
|
||||||
|
return makeDenyReason({
|
||||||
|
code: 'invalid_resource',
|
||||||
|
stage: 'resource_allowlist',
|
||||||
|
message: 'Requested federation resource is not supported',
|
||||||
|
context,
|
||||||
|
resource,
|
||||||
|
details: { supportedResources: FEDERATION_RESOURCE_VALUES },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if (scope.excluded_resources.includes(resource)) {
|
||||||
|
return makeDenyReason({
|
||||||
|
code: 'resource_excluded',
|
||||||
|
stage: 'resource_exclusion',
|
||||||
|
message: 'Requested federation resource is explicitly excluded by grant scope',
|
||||||
|
context,
|
||||||
|
resource,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!scope.resources.includes(resource)) {
|
||||||
|
return makeDenyReason({
|
||||||
|
code: 'resource_not_granted',
|
||||||
|
stage: 'resource_allowlist',
|
||||||
|
message: 'Requested federation resource is not granted by scope',
|
||||||
|
context,
|
||||||
|
resource,
|
||||||
|
details: { grantedResources: scope.resources },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if (requestedLimit !== undefined && (!Number.isInteger(requestedLimit) || requestedLimit < 1)) {
|
||||||
|
return makeDenyReason({
|
||||||
|
code: 'invalid_limit',
|
||||||
|
stage: 'row_cap',
|
||||||
|
statusCode: 400,
|
||||||
|
message: 'Requested row limit must be a positive integer',
|
||||||
|
context,
|
||||||
|
resource,
|
||||||
|
details: { requestedLimit },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
const nativeResult = await nativeRbac.evaluateReadAccess({
|
||||||
|
grantId: context.grantId,
|
||||||
|
peerId: context.peerId,
|
||||||
|
subjectUserId: context.subjectUserId,
|
||||||
|
resource,
|
||||||
|
});
|
||||||
|
|
||||||
|
if (!nativeResult.allowed) {
|
||||||
|
return makeDenyReason({
|
||||||
|
code: 'native_rbac_denied',
|
||||||
|
stage: 'native_rbac',
|
||||||
|
message: nativeResult.reason ?? 'Subject user is not allowed to read this resource',
|
||||||
|
context,
|
||||||
|
resource,
|
||||||
|
...(nativeResult.details !== undefined ? { details: nativeResult.details } : {}),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
const scopeFilter = scope.filters?.[resource];
|
||||||
|
const includePersonal =
|
||||||
|
Boolean(scopeFilter?.include_personal ?? true) && nativeResult.access.includePersonal;
|
||||||
|
const teamIds = intersectTeamIds(nativeResult.access.teamIds, scopeFilter?.include_teams);
|
||||||
|
const limit = Math.min(requestedLimit ?? scope.max_rows_per_query, scope.max_rows_per_query);
|
||||||
|
|
||||||
|
return {
|
||||||
|
allowed: true,
|
||||||
|
filter: {
|
||||||
|
resource,
|
||||||
|
subjectUserId: context.subjectUserId,
|
||||||
|
includePersonal,
|
||||||
|
teamIds,
|
||||||
|
limit,
|
||||||
|
maxRowsPerQuery: scope.max_rows_per_query,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
60
docs/scratchpads/462-fed-m3-04-scope-service.md
Normal file
60
docs/scratchpads/462-fed-m3-04-scope-service.md
Normal file
@@ -0,0 +1,60 @@
|
|||||||
|
# Scratchpad — FED-M3-04 Scope Service
|
||||||
|
|
||||||
|
## Objective
|
||||||
|
|
||||||
|
Implement `apps/gateway/src/federation/server/scope.service.ts` for the M3 inbound federation scope-enforcement pipeline.
|
||||||
|
|
||||||
|
## Scope / Constraints
|
||||||
|
|
||||||
|
- Task: FED-M3-04, issue #462.
|
||||||
|
- Branch: `feat/federation-m3-scope-service` from `origin/main` @ 0.0.48.
|
||||||
|
- Pure service: no direct DB access; native RBAC/data access is injected per evaluation call.
|
||||||
|
- Reuse `parseFederationScope` from M2-03.
|
||||||
|
- Workers do not edit `docs/federation/TASKS.md` per repo AGENTS.md.
|
||||||
|
|
||||||
|
## Acceptance Criteria
|
||||||
|
|
||||||
|
1. Resource allowlist and `excluded_resources` enforced.
|
||||||
|
2. Native RBAC evaluated as `subjectUserId` through an injected evaluator.
|
||||||
|
3. Scope filter intersection supports `include_teams` and `include_personal` without widening native RBAC.
|
||||||
|
4. `max_rows_per_query` caps requested limits.
|
||||||
|
5. Service returns `{ allowed: true, filter }` or a structured deny reason usable by M4 audit.
|
||||||
|
6. Unit tests cover every deny path.
|
||||||
|
|
||||||
|
## Plan
|
||||||
|
|
||||||
|
1. Inspect existing federation scope/schema/auth guard contracts.
|
||||||
|
2. Add pure `FederationScopeService` plus typed result/filter/deny interfaces.
|
||||||
|
3. Add focused unit tests for happy paths, filter intersection, row cap, and deny paths.
|
||||||
|
4. Export/register service for future verb controllers.
|
||||||
|
5. Run situational tests, baseline gates, code review, then PR.
|
||||||
|
|
||||||
|
## Budget
|
||||||
|
|
||||||
|
- Provided model tier: sonnet.
|
||||||
|
- Estimate from task row: 10K tokens.
|
||||||
|
- Working cap assumption: keep implementation focused to FED-M3-04 surfaces only.
|
||||||
|
|
||||||
|
## Progress
|
||||||
|
|
||||||
|
- Intake complete; dirty base worktree avoided by creating isolated worktree at `/home/jarvis/src/mosaic-mono-v1-fed-m3-04`.
|
||||||
|
- Project PRD and federation task spec reviewed.
|
||||||
|
- Added `FederationScopeService` with structured allow/deny result types and injected native RBAC evaluator contract.
|
||||||
|
- Added unit coverage for happy path, row cap, filter intersection, and every deny path.
|
||||||
|
- Exported/registered the service for upcoming M3 verb controllers.
|
||||||
|
|
||||||
|
## Verification Evidence
|
||||||
|
|
||||||
|
- `pnpm --filter @mosaicstack/gateway test -- src/federation/server/__tests__/scope.service.spec.ts` — pass (10 tests before review update; 11 tests after adding include_personal no-leak coverage).
|
||||||
|
- `pnpm build` — pass (23 successful tasks).
|
||||||
|
- `pnpm typecheck` — pass (41 successful tasks; re-run after review update).
|
||||||
|
- `pnpm lint` — pass (23 successful tasks; re-run after review update).
|
||||||
|
- `pnpm format:check` — pass (re-run after review update).
|
||||||
|
- `pnpm test` — pass after starting local `postgres`/`valkey` and running `pnpm --filter @mosaicstack/db db:push` for the DB-backed cross-user isolation suite (41 successful tasks; gateway 477 passed / 11 skipped).
|
||||||
|
- Code review: `~/.config/mosaic/tools/codex/codex-code-review.sh --uncommitted` — approve, 0 findings.
|
||||||
|
- Security review: `~/.config/mosaic/tools/codex/codex-security-review.sh --uncommitted` — risk none, 0 findings.
|
||||||
|
|
||||||
|
## Risks / Blockers
|
||||||
|
|
||||||
|
- Issue #462 is already closed in provider output; likely milestone tracking mismatch. Will still reference #462 in PR body unless orchestrator redirects.
|
||||||
|
- Local full-test setup required `docker compose up -d postgres valkey` + `db:push`; containers were stopped with `docker compose down` after verification.
|
||||||
67
docs/scratchpads/FED-M3-09-query-source.md
Normal file
67
docs/scratchpads/FED-M3-09-query-source.md
Normal file
@@ -0,0 +1,67 @@
|
|||||||
|
# FED-M3-09 — Query Source Service Scratchpad
|
||||||
|
|
||||||
|
## Objective
|
||||||
|
|
||||||
|
Implement `apps/gateway/src/federation/client/query-source.service.ts` for `source: "local" | "federated:<host>" | "all"` routing.
|
||||||
|
|
||||||
|
## Scope
|
||||||
|
|
||||||
|
- Add QuerySourceService in gateway federation client layer.
|
||||||
|
- Unit-test local-only, single federated peer, all-source fan-out/merge, and per-peer partial failures.
|
||||||
|
- Keep `docs/federation/TASKS.md` read-only per project agent guidance.
|
||||||
|
|
||||||
|
## Constraints / assumptions
|
||||||
|
|
||||||
|
- Issue: #462.
|
||||||
|
- Branch: `feat/federation-m3-query-source` from `origin/main` (`e0e7be70`).
|
||||||
|
- ASSUMPTION: `federated:<host>` should match active outbound peers by `commonName` first and by `endpointUrl` host/hostname as compatibility fallback; source tags use `peer.commonName` per `@mosaicstack/types` source-tag docs.
|
||||||
|
- ASSUMPTION: QuerySourceService provides list/fan-out behavior; get/source routing can be layered later because card acceptance says merge rows.
|
||||||
|
- ASSUMPTION: `source: "all"` cannot safely return a single continuation cursor for multiple sub-sources; any subquery cursor marks the merged response `_partial: true` + `_truncated: true` while omitting `nextCursor`.
|
||||||
|
- Budget: no explicit hard cap from orchestrator; working cap ~8K-12K tokens for card 1 implementation + tests + PR cycle.
|
||||||
|
- OpenBrain unavailable: credential loader failed with missing `/home/jarvis/.config/mosaic/credentials.json`; not blocking code delivery.
|
||||||
|
|
||||||
|
## Plan
|
||||||
|
|
||||||
|
1. Review federation client/types/db patterns.
|
||||||
|
2. Write unit tests for source behavior.
|
||||||
|
3. Implement QuerySourceService and export/register it in FederationModule.
|
||||||
|
4. Run scoped tests, typecheck, lint, format.
|
||||||
|
5. Run codex uncommitted review and remediate.
|
||||||
|
6. Commit, queue guard, push, PR via wrapper.
|
||||||
|
|
||||||
|
## Progress
|
||||||
|
|
||||||
|
- 2026-06-24: Intake complete; using isolated worktree to avoid dirty orchestrator files in original checkout.
|
||||||
|
- 2026-06-24: Added QuerySourceService, module export, barrel export, and 7 unit tests.
|
||||||
|
- 2026-06-24: First Codex review found pagination and port-host matching issues; both remediated with tests.
|
||||||
|
|
||||||
|
## Tests run
|
||||||
|
|
||||||
|
- `pnpm --filter @mosaicstack/gateway test -- query-source.service.spec.ts` — PASS (7 tests).
|
||||||
|
- `pnpm --filter @mosaicstack/gateway typecheck` — PASS.
|
||||||
|
- `pnpm --filter @mosaicstack/gateway lint` — PASS.
|
||||||
|
- `pnpm format:check` — PASS.
|
||||||
|
- `pnpm typecheck` — PASS (41/41 turbo tasks).
|
||||||
|
- `pnpm lint` — PASS (23/23 turbo tasks).
|
||||||
|
- `pnpm test` — FAIL in pre-existing/live-DB integration suite: `apps/gateway/src/__tests__/cross-user-isolation.test.ts` cleanup hit `relation "messages" does not exist` against local PostgreSQL. Changed QuerySource unit tests passed; failure is outside FED-M3-09 surface and appears tied to local DB schema state.
|
||||||
|
|
||||||
|
## Review evidence
|
||||||
|
|
||||||
|
- `~/.config/mosaic/tools/codex/codex-code-review.sh --uncommitted` — first pass request-changes, 2 should-fix findings (all-source cursor handling; endpoint port host matching).
|
||||||
|
- Remediation: `_partial` + `_truncated` when any all-source subquery has `nextCursor`; endpoint match accepts URL `host` and `hostname`; added tests for both.
|
||||||
|
- `~/.config/mosaic/tools/codex/codex-code-review.sh --uncommitted` — PASS/approve, no findings.
|
||||||
|
- `~/.config/mosaic/tools/codex/codex-security-review.sh --uncommitted` — PASS, risk level none, no findings.
|
||||||
|
|
||||||
|
## Risks / blockers
|
||||||
|
|
||||||
|
- Federation query layer is not yet wired; service API needs to be stable and easy to compose.
|
||||||
|
- Must avoid hard-failing `source: all` on remote peer failures.
|
||||||
|
|
||||||
|
## Acceptance evidence mapping
|
||||||
|
|
||||||
|
| Acceptance criterion | Evidence |
|
||||||
|
| ------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- |
|
||||||
|
| local source returns local rows tagged `_source: local` | `query-source.service.spec.ts` local test; scoped test PASS |
|
||||||
|
| `federated:<host>` queries selected peer and tags rows with peer source | `query-source.service.spec.ts` commonName/endpoint-host tests; scoped test PASS |
|
||||||
|
| `all` fans out local + active outbound peers in parallel and merges tagged rows | `query-source.service.spec.ts` all-source call-order/merge test; scoped test PASS |
|
||||||
|
| per-peer failure on `all` returns `_partial: true`, not throw | `query-source.service.spec.ts` peer failure test; scoped test PASS |
|
||||||
Reference in New Issue
Block a user