342 lines
12 KiB
TypeScript
342 lines
12 KiB
TypeScript
import { mkdtemp, rm } from 'node:fs/promises';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
|
|
import { Test, type TestingModule } from '@nestjs/testing';
|
|
import {
|
|
connectorLeaseAuditLog,
|
|
createPgliteDb,
|
|
eq,
|
|
runPgliteMigrations,
|
|
type DbHandle,
|
|
} from '@mosaicstack/db';
|
|
import type { ConnectorExecutionContext, FencedConnectorAdapter } from '@mosaicstack/types';
|
|
import { DB } from '../database/database.module.js';
|
|
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
|
|
import {
|
|
CONNECTOR_LEASE_POLICY,
|
|
ConnectorLeaseService,
|
|
type ConnectorLeasePolicy,
|
|
type ConnectorLeasePolicySubject,
|
|
} from './connector-lease.service.js';
|
|
|
|
const authorize = vi.fn().mockResolvedValue(true);
|
|
const policy: ConnectorLeasePolicy = { authorize };
|
|
const context = {
|
|
actorScope: { userId: 'operator-a', tenantId: 'tenant-a' },
|
|
correlationId: 'correlation-acquire',
|
|
};
|
|
|
|
describe('gateway connector lease fencing integration', (): void => {
|
|
let dataDir: string;
|
|
let handle: DbHandle;
|
|
let moduleRef: TestingModule;
|
|
let service: ConnectorLeaseService;
|
|
let repository: ConnectorLeaseRepository;
|
|
|
|
beforeAll(async (): Promise<void> => {
|
|
vi.useFakeTimers();
|
|
vi.setSystemTime(new Date('2026-07-14T17:00:00.000Z'));
|
|
dataDir = await mkdtemp(join(tmpdir(), 'mosaic-gateway-connector-lease-'));
|
|
handle = createPgliteDb(dataDir);
|
|
await runPgliteMigrations(handle);
|
|
moduleRef = await Test.createTestingModule({
|
|
providers: [
|
|
ConnectorLeaseRepository,
|
|
ConnectorLeaseService,
|
|
{ provide: DB, useValue: handle.db },
|
|
{ provide: CONNECTOR_LEASE_POLICY, useValue: policy },
|
|
],
|
|
}).compile();
|
|
service = moduleRef.get(ConnectorLeaseService);
|
|
repository = moduleRef.get(ConnectorLeaseRepository);
|
|
});
|
|
|
|
afterAll(async (): Promise<void> => {
|
|
vi.useRealTimers();
|
|
await moduleRef.close();
|
|
await handle.close();
|
|
await rm(dataDir, { recursive: true, force: true });
|
|
});
|
|
|
|
it('derives tenant authority at the gateway and validates a grant before side effects', async (): Promise<void> => {
|
|
const lease = await service.acquire(
|
|
{
|
|
logicalAgentId: 'Mos',
|
|
bindingId: 'operator-chat',
|
|
connectorId: 'pi-worker-a',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
},
|
|
context,
|
|
);
|
|
const grant = await service.issueGrant(
|
|
{ lease, scopes: ['runtime.send'], ttlMs: 30_000 },
|
|
{ ...context, correlationId: 'correlation-grant' },
|
|
);
|
|
const execute = vi.fn(async (_message: string, leaseContext: ConnectorExecutionContext) => {
|
|
return leaseContext.leaseEpoch;
|
|
});
|
|
const adapter: FencedConnectorAdapter<string, string> = { execute };
|
|
|
|
await expect(service.executeGrant(grant, 'runtime.send', 'hello', adapter)).resolves.toBe('1');
|
|
expect(execute).toHaveBeenCalledOnce();
|
|
expect(authorize).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
action: 'grant.issue',
|
|
requestedScopes: ['runtime.send'],
|
|
requestedTtlMs: 30_000,
|
|
}),
|
|
);
|
|
expect(execute.mock.calls[0]?.[1]).toMatchObject({
|
|
identity: { tenantId: 'tenant-a', logicalAgentId: 'mos' },
|
|
bindingId: 'operator-chat',
|
|
connectorId: 'pi-worker-a',
|
|
});
|
|
});
|
|
|
|
it('normalizes lease-derived policy subjects before authorization', async (): Promise<void> => {
|
|
const lease = await service.acquire(
|
|
{
|
|
logicalAgentId: 'mos',
|
|
bindingId: 'operator-chat-policy',
|
|
connectorId: 'pi-worker-a',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
},
|
|
{ ...context, correlationId: 'correlation-policy-setup' },
|
|
);
|
|
const aliasedLease = {
|
|
...lease,
|
|
identity: { ...lease.identity, logicalAgentId: ' MOS ' },
|
|
bindingId: ' Operator-Chat-Policy ',
|
|
connectorId: ' PI-Worker-A ',
|
|
scopes: [' Runtime.Send '],
|
|
leaseEpoch: `00${lease.leaseEpoch}`,
|
|
};
|
|
|
|
await service.heartbeat(aliasedLease, 30_000, {
|
|
...context,
|
|
correlationId: 'correlation-policy-heartbeat',
|
|
});
|
|
expect(authorize).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({
|
|
action: 'lease.heartbeat',
|
|
logicalAgentId: 'mos',
|
|
bindingId: 'operator-chat-policy',
|
|
connectorId: 'pi-worker-a',
|
|
requestedScopes: ['runtime.send'],
|
|
}),
|
|
);
|
|
|
|
await service.issueGrant(
|
|
{ lease: aliasedLease, scopes: [' Runtime.Send '], ttlMs: 1_000 },
|
|
{ ...context, correlationId: 'correlation-policy-grant' },
|
|
);
|
|
expect(authorize).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({
|
|
action: 'grant.issue',
|
|
logicalAgentId: 'mos',
|
|
bindingId: 'operator-chat-policy',
|
|
connectorId: 'pi-worker-a',
|
|
requestedScopes: ['runtime.send'],
|
|
}),
|
|
);
|
|
|
|
await service.release(aliasedLease, {
|
|
...context,
|
|
correlationId: 'correlation-policy-release',
|
|
});
|
|
expect(authorize).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({
|
|
action: 'lease.release',
|
|
logicalAgentId: 'mos',
|
|
bindingId: 'operator-chat-policy',
|
|
connectorId: 'pi-worker-a',
|
|
requestedScopes: ['runtime.send'],
|
|
}),
|
|
);
|
|
});
|
|
|
|
it('denies stale, forged, expired, cross-tenant, and cross-binding grants before effects', async (): Promise<void> => {
|
|
const bindingId = 'operator-chat-denials';
|
|
const current = await service.acquire(
|
|
{
|
|
logicalAgentId: 'mos',
|
|
bindingId,
|
|
connectorId: 'pi-worker-a',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
},
|
|
{ ...context, correlationId: 'correlation-denial-setup' },
|
|
);
|
|
const stale = await service.issueGrant(
|
|
{ lease: current, scopes: ['runtime.send'], ttlMs: 30_000 },
|
|
{ ...context, correlationId: 'correlation-stale' },
|
|
);
|
|
await service.takeover(
|
|
{
|
|
logicalAgentId: 'mos',
|
|
bindingId,
|
|
connectorId: 'pi-worker-b',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
expectedEpoch: current.leaseEpoch,
|
|
},
|
|
{ ...context, correlationId: 'correlation-takeover' },
|
|
);
|
|
const adapter = { execute: vi.fn().mockResolvedValue(undefined) };
|
|
|
|
await expect(service.executeGrant(stale, 'runtime.send', undefined, adapter)).rejects.toThrow();
|
|
|
|
const active = await service.current('mos', bindingId, context);
|
|
if (!active) throw new Error('active lease fixture is unavailable');
|
|
const grant = await service.issueGrant(
|
|
{ lease: active, scopes: ['runtime.send'], ttlMs: 1_000 },
|
|
{ ...context, correlationId: 'correlation-active' },
|
|
);
|
|
await expect(
|
|
service.executeGrant({ ...grant }, 'runtime.send', undefined, adapter),
|
|
).rejects.toThrow();
|
|
await expect(
|
|
service.executeGrant(
|
|
{ ...grant, bindingId: 'other-binding' },
|
|
'runtime.send',
|
|
undefined,
|
|
adapter,
|
|
),
|
|
).rejects.toThrow();
|
|
await expect(
|
|
service.issueGrant(
|
|
{ lease: active, scopes: ['runtime.send'], ttlMs: 30_000 },
|
|
{
|
|
actorScope: { userId: 'operator-b', tenantId: 'tenant-b' },
|
|
correlationId: 'correlation-cross-tenant',
|
|
},
|
|
),
|
|
).rejects.toThrow();
|
|
const crossTenantAudit = await handle.db
|
|
.select()
|
|
.from(connectorLeaseAuditLog)
|
|
.where(eq(connectorLeaseAuditLog.correlationId, 'correlation-cross-tenant'));
|
|
expect(crossTenantAudit).toHaveLength(1);
|
|
expect(crossTenantAudit[0]).toMatchObject({
|
|
tenantId: 'tenant-b',
|
|
logicalAgentId: 'untrusted',
|
|
bindingId: 'untrusted',
|
|
connectorId: 'untrusted',
|
|
reason: 'policy_denied',
|
|
});
|
|
|
|
vi.setSystemTime(new Date('2026-07-14T17:00:02.000Z'));
|
|
await expect(service.executeGrant(grant, 'runtime.send', undefined, adapter)).rejects.toThrow();
|
|
expect(adapter.execute).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('rejects submitted lifecycle scopes that differ from durable authority before policy or mutation', async (): Promise<void> => {
|
|
authorize.mockResolvedValue(true);
|
|
const heartbeatLease = await service.acquire(
|
|
{
|
|
logicalAgentId: 'mos',
|
|
bindingId: 'operator-chat-heartbeat-scope',
|
|
connectorId: 'pi-worker-a',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
},
|
|
{ ...context, correlationId: 'correlation-heartbeat-scope-setup' },
|
|
);
|
|
const releaseLease = await service.acquire(
|
|
{
|
|
logicalAgentId: 'mos',
|
|
bindingId: 'operator-chat-release-scope',
|
|
connectorId: 'pi-worker-a',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
},
|
|
{ ...context, correlationId: 'correlation-release-scope-setup' },
|
|
);
|
|
const forgedHeartbeat = { ...heartbeatLease, scopes: ['tool.execute'] };
|
|
const forgedRelease = { ...releaseLease, scopes: ['tool.execute'] };
|
|
|
|
authorize.mockImplementation(async (subject: ConnectorLeasePolicySubject) => {
|
|
return subject.requestedScopes.length === 1 && subject.requestedScopes[0] === 'tool.execute';
|
|
});
|
|
authorize.mockClear();
|
|
|
|
await expect(
|
|
service.heartbeat(forgedHeartbeat, 30_000, {
|
|
...context,
|
|
correlationId: 'correlation-heartbeat-scope-forgery',
|
|
}),
|
|
).rejects.toThrow('Connector authority policy denied');
|
|
await expect(
|
|
service.release(forgedRelease, {
|
|
...context,
|
|
correlationId: 'correlation-release-scope-forgery',
|
|
}),
|
|
).rejects.toThrow('Connector authority policy denied');
|
|
expect(authorize).not.toHaveBeenCalled();
|
|
|
|
const currentHeartbeat = await repository.findCurrent({
|
|
identity: heartbeatLease.identity,
|
|
bindingId: heartbeatLease.bindingId,
|
|
});
|
|
const currentRelease = await repository.findCurrent({
|
|
identity: releaseLease.identity,
|
|
bindingId: releaseLease.bindingId,
|
|
});
|
|
expect(currentHeartbeat).toMatchObject({
|
|
leaseId: heartbeatLease.leaseId,
|
|
scopes: ['runtime.send'],
|
|
heartbeatAt: heartbeatLease.heartbeatAt,
|
|
expiresAt: heartbeatLease.expiresAt,
|
|
});
|
|
expect(currentRelease).toMatchObject({
|
|
leaseId: releaseLease.leaseId,
|
|
scopes: ['runtime.send'],
|
|
});
|
|
expect(currentRelease?.releasedAt).toBeUndefined();
|
|
|
|
const forgedAudits = await handle.db
|
|
.select()
|
|
.from(connectorLeaseAuditLog)
|
|
.where(eq(connectorLeaseAuditLog.correlationId, 'correlation-heartbeat-scope-forgery'));
|
|
expect(forgedAudits).toHaveLength(1);
|
|
expect(forgedAudits[0]).toMatchObject({
|
|
bindingId: heartbeatLease.bindingId,
|
|
connectorId: heartbeatLease.connectorId,
|
|
event: 'reject',
|
|
outcome: 'denied',
|
|
reason: 'policy_denied',
|
|
});
|
|
const forgedReleaseAudits = await handle.db
|
|
.select()
|
|
.from(connectorLeaseAuditLog)
|
|
.where(eq(connectorLeaseAuditLog.correlationId, 'correlation-release-scope-forgery'));
|
|
expect(forgedReleaseAudits).toHaveLength(1);
|
|
expect(forgedReleaseAudits[0]).toMatchObject({
|
|
bindingId: releaseLease.bindingId,
|
|
connectorId: releaseLease.connectorId,
|
|
event: 'reject',
|
|
outcome: 'denied',
|
|
reason: 'policy_denied',
|
|
});
|
|
|
|
authorize.mockImplementation(async (subject: ConnectorLeasePolicySubject) => {
|
|
return subject.requestedScopes.length === 1 && subject.requestedScopes[0] === 'runtime.send';
|
|
});
|
|
await expect(
|
|
service.heartbeat(heartbeatLease, 30_000, {
|
|
...context,
|
|
correlationId: 'correlation-heartbeat-scope-canonical',
|
|
}),
|
|
).resolves.toMatchObject({ scopes: ['runtime.send'] });
|
|
await expect(
|
|
service.release(releaseLease, {
|
|
...context,
|
|
correlationId: 'correlation-release-scope-canonical',
|
|
}),
|
|
).resolves.toBeUndefined();
|
|
});
|
|
});
|