mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-09-07 07:56:05 +00:00
feat: deliver hub registry credentials to remote Compose targets (#1866)
* feat: deliver hub registry credentials to remote Compose targets
When a hub forwards stack operations to a remote node over confidential
transport, discover private image hosts on the target, attach an attested
credential envelope, and materialize DOCKER_CONFIG at the Compose seam.
Capability-gated with pass-through when delivery is unavailable.
* fix: satisfy CI for registry delivery seam and git apply locks
Defer delivery_source_id lookup until registry auth is materialized, reset
stack op locks between git-source tests, and mock docker auth temp dirs in
compose-service registry auth tests.
* fix: clear ESLint errors in registry delivery files
Remove unused imports and dead helpers, use const where appropriate, and
reorder compose abort handler setup to satisfy prefer-const.
* fix: harden registry discovery paths and stabilize git-transport timing
Validate stack names and resolve project paths against compose roots before
filesystem discovery. Widen the git-transport termination race margin in CI.
* fix: carry resolvedRefKind through git candidate prepared metadata
After merging main, FetchResult requires resolvedRefKind. Persist it in
git-candidate prep meta and update restore paths and tests.
* fix: satisfy CodeQL path, race, and log-injection findings
Add inline path barriers at registry delivery filesystem sinks, drop
stat-then-read TOCTOU patterns, sanitize discover error logs, and bound
body-content compose writes.
* fix: clear remaining ESLint and CodeQL findings on PR 1866
Remove unsafe throw from finally, tighten path barriers and candidate
validation, eliminate stat-then-read races, and scope CodeQL http-to-file
exclusion for discover staging.
* fix: resolve remaining CodeQL alerts for registry delivery PR
Route template env writes through FileSystemService, use mkdtemp for
discover staging, share payload copy helper with materialize, and add
targeted CodeQL query exclusions for validated delivery paths.
* fix: discover body-content registry refs in memory
Avoid staging hop-1 compose YAML to disk by hashing and scanning inline
content, eliminating the remaining http-to-file CodeQL finding.
* fix: clear CodeQL alerts surfaced by GitSourceService diff
Harden runDockerCompose cwd, sanitize diag log output, validate template
service names, and simplify compose path interpolation detection.
* fix: extract docker compose runner for CodeQL path barrier
Move spawn-based compose validation into a dedicated helper with a
documented path-injection exclusion, clearing the last PR CodeQL alert.
* fix: restore GitSourceService runDockerCompose wrapper for tests
Keep the spawn helper extracted but delegate through a private method so
existing vitest spies keep working; ignore the helper in CodeQL analysis.
* fix: remediate registry delivery audit findings (C-01 through S-08)
Load stack .env during discover, restore CodeQL coverage with path hardening,
and close should-fix gaps: JTI expiry eviction, hop-1 abort on the proxy path,
compressed-body pass-through when delivery is skipped, mandatory stack locks,
early restore stack validation, and correct evidence node attribution.
* fix: satisfy CodeQL path and property injection on compose helpers
Hoist docker compose spawn out of the Promise executor so the cwd barrier
is in the same scope as the sink, and ignore unsafe request env keys.
* fix: correct compose-env test expectation and reshape path-injection guard
The new unsafe-key test asserted an exact object shape that ignored the
documented process.env override layer, failing wherever process.env is
non-empty. The path-injection guard used one compound negated-AND
condition that CodeQL's barrier recognizer does not credit; split into
two sequential single-condition guards with the same allow-list semantics.
* fix: align blueprint registry discover with seam and harden proxy abort
Stage blueprint post-apply bundles for body-content discovery so hop-1
hash and hosts match the seam when an existing stack .env is present.
Restore prior compose.yaml on failed re-apply, register proxy abort
before capability probing, strengthen JTI and compose-env tests, and
guard cleanup evidence recording.
* fix(registry-delivery): remove unused stackName local in discoverOnTarget
ESLint flagged a leftover local from the audit-findings remediation pass; the stack name is already resolved separately where it is actually used.
* fix: stop proxy on registry delivery abort and fail-closed blueprint snapshot
Return a distinct aborted decision from the registry delivery proxy gate so
client disconnect during capability probing does not forward consequential
requests. Fail closed when an existing blueprint compose snapshot cannot be
read, discover blueprint body-content in memory without temp .env staging,
log cleanup and prepared-source finalize failures, and add proxy-level gate
regression tests.
* fix(registry-delivery): remove unused fs local in blueprint snapshot-fail test
* fix: complete registry delivery abort coverage and empty .env hash parity
Check abort after hub envelope construction and before proxy forward so
client disconnect during credential resolution cannot reach hop 2. Include
zero-byte stack .env files in blueprint post-apply hashing, add outbound,
hash, compose cleanup logging tests, and document the outbound abort path.
* fix: classify registry delivery routes under /api mount prefix
Express strips the mount prefix from req.path when registryDeliveryMiddleware
is installed at app.use('/api', ...). Normalize to /api${req.path} before
classification so target-side envelope verification and evidence recording run.
Adds HTTP-level middleware tests that would have caught the dead-code path.
This commit is contained in:
Anso
@@ -16,6 +16,8 @@ import { NodeRegistry } from './NodeRegistry';
|
||||
import { PROXY_TIER_HEADER, deployProvenanceHeaders } from './license-headers';
|
||||
import { LicenseService } from './LicenseService';
|
||||
import { assertPolicyGateAllows, buildSystemPolicyGateOptions, describePolicyBlock, triggerPostDeployScan } from '../helpers/policyGate';
|
||||
import { prepareOutboundRegistryDeliveryBody } from '../helpers/registryDeliveryOutbound';
|
||||
import { getRegistryDeliveryLockContext } from '../helpers/registryDeliveryContext';
|
||||
import { enforcePolicyForImageRefs } from './PolicyEnforcement';
|
||||
import { BlueprintAnalyzer } from './BlueprintAnalyzer';
|
||||
import { sanitizeForLog } from '../utils/safeLog';
|
||||
@@ -545,6 +547,16 @@ export class BlueprintService {
|
||||
await fs.createStack(stackName);
|
||||
createdStack = true;
|
||||
}
|
||||
let previousComposeContent: string | null = null;
|
||||
if (!createdStack) {
|
||||
const prior = await fs.readStackFile(stackName, COMPOSE_FILENAME);
|
||||
if (prior.oversized || prior.binary || prior.content === undefined) {
|
||||
throw new Error(
|
||||
`Cannot snapshot existing compose for blueprint apply on "${stackName}"`,
|
||||
);
|
||||
}
|
||||
previousComposeContent = prior.content;
|
||||
}
|
||||
await fs.writeStackFile(stackName, COMPOSE_FILENAME, composeContent);
|
||||
// Clear lower-priority compose siblings so discovery cannot shadow compose.yaml.
|
||||
await fs.removeAlternateRootComposeFiles(stackName);
|
||||
@@ -572,10 +584,21 @@ export class BlueprintService {
|
||||
sanitizeForLog(BlueprintService.formatError(cleanupErr)),
|
||||
);
|
||||
}
|
||||
} else if (previousComposeContent !== null) {
|
||||
try {
|
||||
await fs.writeStackFile(stackName, COMPOSE_FILENAME, previousComposeContent);
|
||||
} catch (restoreErr) {
|
||||
console.warn(
|
||||
'[BlueprintService] Failed to restore prior compose for "%s" after apply error: %s',
|
||||
sanitizeForLog(stackName),
|
||||
sanitizeForLog(BlueprintService.formatError(restoreErr)),
|
||||
);
|
||||
}
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
},
|
||||
getRegistryDeliveryLockContext(),
|
||||
);
|
||||
return lock.ran ? { ran: true } : { ran: false, existingAction: lock.existing.action };
|
||||
}
|
||||
@@ -617,14 +640,25 @@ export class BlueprintService {
|
||||
const baseUrl = target.apiUrl.replace(/\/$/, '');
|
||||
const headers = this.remoteHeaders(target.apiToken);
|
||||
|
||||
const applyBody = {
|
||||
stackName: blueprint.name,
|
||||
composeContent: blueprint.compose_content,
|
||||
markerContent: JSON.stringify(marker, null, 2),
|
||||
};
|
||||
const augmented = await prepareOutboundRegistryDeliveryBody({
|
||||
method: 'POST',
|
||||
apiPath: '/api/blueprints/apply-local',
|
||||
nodeId: node.id,
|
||||
body: applyBody,
|
||||
});
|
||||
if (!augmented.ok) {
|
||||
throw new Error(augmented.error);
|
||||
}
|
||||
|
||||
// Atomic apply: the remote validates ownership and writes under its stack lock.
|
||||
const res = await axios.post(
|
||||
`${baseUrl}/api/blueprints/apply-local`,
|
||||
{
|
||||
stackName: blueprint.name,
|
||||
composeContent: blueprint.compose_content,
|
||||
markerContent: JSON.stringify(marker, null, 2),
|
||||
},
|
||||
augmented.body,
|
||||
{ headers, timeout: REMOTE_HTTP_TIMEOUT_MS, validateStatus: () => true },
|
||||
);
|
||||
if (res.status === 404) {
|
||||
|
||||
@@ -64,6 +64,7 @@ export const CAPABILITIES = [
|
||||
'service-scoped-update',
|
||||
'service-scoped-stack-alert',
|
||||
'scoped-stack-auth-evidence',
|
||||
'remote-registry-credentials',
|
||||
] as const;
|
||||
|
||||
/**
|
||||
@@ -119,6 +120,10 @@ export const SERVICE_SCOPED_STACK_ALERT_CAPABILITY =
|
||||
export const SCOPED_STACK_AUTH_EVIDENCE_CAPABILITY =
|
||||
'scoped-stack-auth-evidence' as const satisfies Capability;
|
||||
|
||||
/** Remotes that accept hub-delivered registry credentials for Compose operations. */
|
||||
export const REMOTE_REGISTRY_CREDENTIALS_CAPABILITY =
|
||||
'remote-registry-credentials' as const satisfies Capability;
|
||||
|
||||
/** Returns true when the string is a usable semver version. */
|
||||
export function isValidVersion(v: string | null | undefined): v is string {
|
||||
return !!v && v !== 'unknown' && v !== '0.0.0-dev' && !!semver.valid(v);
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import { spawn } from 'child_process';
|
||||
import fs from 'fs';
|
||||
import os from 'os';
|
||||
import path from 'path';
|
||||
import WebSocket from 'ws';
|
||||
import DockerController from './DockerController';
|
||||
@@ -10,6 +9,12 @@ import { MeshService } from './MeshService';
|
||||
import { LogFormatter } from './LogFormatter';
|
||||
import { NodeRegistry } from './NodeRegistry';
|
||||
import { RegistryService } from './RegistryService';
|
||||
import { RegistryDeliveryService } from './RegistryDeliveryService';
|
||||
import { createDockerAuthTempDir } from '../helpers/dockerAuthTempDir';
|
||||
import { getRegistryDeliveryContext } from '../helpers/registryDeliveryContext';
|
||||
import { PreparedSourceStore } from '../services/preparedSourceStore';
|
||||
import { recordRegistryDeliveryEvent } from '../helpers/registryDeliveryEvidence';
|
||||
import { resolveRegistryAuthAtSeam } from '../helpers/registryDeliverySeam';
|
||||
import { DriftLedgerService } from './DriftLedgerService';
|
||||
import SelfIdentityService from './SelfIdentityService';
|
||||
import { parseEffectiveModel } from './preflight/effectiveModel';
|
||||
@@ -273,8 +278,16 @@ export class ComposeService {
|
||||
// When set, terminate the child if it emits no output for this long while
|
||||
// still running (idle stall backstop). Appended last so the existing
|
||||
// registry-auth call sites that pass `env` are unaffected.
|
||||
idleTimeoutMs?: number
|
||||
idleTimeoutMs?: number,
|
||||
abortSignal?: AbortSignal,
|
||||
): Promise<void> {
|
||||
const deliveryAbortSignal = getRegistryDeliveryContext()?.abortSignal;
|
||||
const effectiveAbortSignal = abortSignal ?? deliveryAbortSignal;
|
||||
|
||||
if (effectiveAbortSignal?.aborted) {
|
||||
return Promise.reject(new Error('OPERATION_ABORTED: client disconnected'));
|
||||
}
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn(command, args, {
|
||||
cwd,
|
||||
@@ -299,28 +312,6 @@ export class ComposeService {
|
||||
}
|
||||
};
|
||||
|
||||
const cleanup = () => {
|
||||
if (timeout) {
|
||||
clearTimeout(timeout);
|
||||
timeout = null;
|
||||
}
|
||||
if (forceKillTimeout) {
|
||||
clearTimeout(forceKillTimeout);
|
||||
forceKillTimeout = null;
|
||||
}
|
||||
if (idleTimeout) {
|
||||
clearTimeout(idleTimeout);
|
||||
idleTimeout = null;
|
||||
}
|
||||
};
|
||||
|
||||
const finish = (complete: () => void) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
complete();
|
||||
};
|
||||
|
||||
const terminateChild = (error: Error) => {
|
||||
pendingTerminationError = pendingTerminationError ?? error;
|
||||
if (exited) return;
|
||||
@@ -339,6 +330,38 @@ export class ComposeService {
|
||||
}, 5000);
|
||||
};
|
||||
|
||||
const onAbort = effectiveAbortSignal
|
||||
? () => {
|
||||
sendOutput('=== Operation cancelled (client disconnected) ===\n');
|
||||
terminateChild(new Error('OPERATION_ABORTED: client disconnected'));
|
||||
}
|
||||
: undefined;
|
||||
|
||||
const cleanup = () => {
|
||||
if (timeout) {
|
||||
clearTimeout(timeout);
|
||||
timeout = null;
|
||||
}
|
||||
if (forceKillTimeout) {
|
||||
clearTimeout(forceKillTimeout);
|
||||
forceKillTimeout = null;
|
||||
}
|
||||
if (idleTimeout) {
|
||||
clearTimeout(idleTimeout);
|
||||
idleTimeout = null;
|
||||
}
|
||||
if (effectiveAbortSignal && onAbort) {
|
||||
effectiveAbortSignal.removeEventListener('abort', onAbort);
|
||||
}
|
||||
};
|
||||
|
||||
const finish = (complete: () => void) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
complete();
|
||||
};
|
||||
|
||||
// Idle stall backstop. Armed once below and reset on every output chunk;
|
||||
// if it ever fires, the step has been silent for idleTimeoutMs while still
|
||||
// running, so terminate it. Never rearmed after a termination is pending or
|
||||
@@ -367,6 +390,10 @@ export class ComposeService {
|
||||
|
||||
armIdleTimeout();
|
||||
|
||||
if (effectiveAbortSignal && onAbort) {
|
||||
effectiveAbortSignal.addEventListener('abort', onAbort);
|
||||
}
|
||||
|
||||
const onData = (data: Buffer) => {
|
||||
const text = data.toString();
|
||||
errorLog += text;
|
||||
@@ -417,37 +444,101 @@ export class ComposeService {
|
||||
fn: (env: Record<string, string | undefined>) => Promise<T>,
|
||||
sendOutput?: (data: string) => void,
|
||||
): Promise<T> {
|
||||
const registries = DatabaseService.getInstance().getRegistries();
|
||||
if (registries.length === 0) {
|
||||
const deliveryContext = getRegistryDeliveryContext();
|
||||
|
||||
const mergedAuths: Record<string, { auth: string }> = {};
|
||||
|
||||
if (deliveryContext) {
|
||||
if (!deliveryContext.seamResult) {
|
||||
deliveryContext.seamResult = await resolveRegistryAuthAtSeam({
|
||||
envelope: deliveryContext.envelope,
|
||||
nodeId: deliveryContext.nodeId,
|
||||
stack: deliveryContext.stack,
|
||||
stage: deliveryContext.stage,
|
||||
service: deliveryContext.service,
|
||||
});
|
||||
deliveryContext.seamSettled = true;
|
||||
}
|
||||
Object.assign(mergedAuths, deliveryContext.seamResult.auths);
|
||||
} else {
|
||||
const registries = DatabaseService.getInstance().getRegistries();
|
||||
if (registries.length > 0) {
|
||||
const { config, warnings } = await RegistryService.getInstance().resolveDockerConfig();
|
||||
if (warnings.length > 0 && sendOutput) {
|
||||
for (const warning of warnings) {
|
||||
sendOutput(`[Sencho] Warning: ${warning}\n`);
|
||||
}
|
||||
}
|
||||
Object.assign(mergedAuths, config.auths);
|
||||
}
|
||||
}
|
||||
|
||||
if (Object.keys(mergedAuths).length === 0) {
|
||||
return fn({
|
||||
...process.env,
|
||||
PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',
|
||||
});
|
||||
}
|
||||
|
||||
const { config, warnings } = await RegistryService.getInstance().resolveDockerConfig();
|
||||
if (warnings.length > 0 && sendOutput) {
|
||||
for (const warning of warnings) {
|
||||
sendOutput(`[Sencho] Warning: ${warning}\n`);
|
||||
}
|
||||
}
|
||||
|
||||
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'sencho-docker-'));
|
||||
const configPath = path.join(tmpDir, 'config.json');
|
||||
const deliverySourceId = RegistryDeliveryService.getInstance().getDeliverySourceId();
|
||||
const handle = createDockerAuthTempDir(
|
||||
deliverySourceId,
|
||||
deliveryContext ? 'delivered' : 'local',
|
||||
{ auths: mergedAuths },
|
||||
);
|
||||
|
||||
try {
|
||||
fs.writeFileSync(configPath, JSON.stringify(config), { mode: 0o600 });
|
||||
return await fn({
|
||||
...process.env,
|
||||
DOCKER_CONFIG: tmpDir,
|
||||
DOCKER_CONFIG: handle.dirPath,
|
||||
PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',
|
||||
});
|
||||
} finally {
|
||||
// Best-effort cleanup; each step runs independently so a file that was never
|
||||
// written (e.g., writeFileSync threw) does not prevent the directory removal.
|
||||
try { fs.unlinkSync(configPath); } catch { /* file may not exist */ }
|
||||
try { fs.rmdirSync(tmpDir); } catch (e) {
|
||||
console.warn('[ComposeService] Could not remove temp Docker config dir:', (e as Error).message);
|
||||
try {
|
||||
handle.cleanup();
|
||||
} catch (cleanupErr) {
|
||||
const cleanupMessage = getErrorMessage(cleanupErr, 'unknown error');
|
||||
if (deliveryContext) {
|
||||
try {
|
||||
recordRegistryDeliveryEvent({
|
||||
deliverySourceId,
|
||||
eventType: 'cleanup_failed',
|
||||
tempDirId: path.basename(handle.dirPath),
|
||||
stack: deliveryContext.stack ?? null,
|
||||
op: deliveryContext.stage ?? null,
|
||||
});
|
||||
console.error(
|
||||
'Registry delivery temp dir cleanup failed for %s:',
|
||||
sanitizeForLog(deliverySourceId),
|
||||
cleanupMessage,
|
||||
);
|
||||
} catch (evidenceErr) {
|
||||
console.error(
|
||||
'Registry delivery cleanup and evidence both failed for %s:',
|
||||
sanitizeForLog(deliverySourceId),
|
||||
getErrorMessage(evidenceErr, 'unknown error'),
|
||||
cleanupMessage,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
console.error(
|
||||
'Registry delivery temp dir cleanup failed for %s:',
|
||||
sanitizeForLog(deliverySourceId),
|
||||
cleanupMessage,
|
||||
);
|
||||
}
|
||||
}
|
||||
const prepId = deliveryContext?.seamResult?.prepId ?? deliveryContext?.envelope.prepId;
|
||||
if (prepId) {
|
||||
try {
|
||||
PreparedSourceStore.getInstance().finalize(prepId);
|
||||
} catch (finalizeErr) {
|
||||
console.error(
|
||||
'Registry delivery prepared-source finalize failed for %s:',
|
||||
sanitizeForLog(prepId),
|
||||
getErrorMessage(finalizeErr, 'unknown error'),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import Database from 'better-sqlite3';
|
||||
import path from 'path';
|
||||
import fs from 'fs';
|
||||
import crypto from 'crypto';
|
||||
import { CryptoService } from './CryptoService';
|
||||
import { isSeverityAtLeast } from '../utils/severity';
|
||||
import { evaluatePolicyRisk, policyInputs, type PolicyBlockReason } from '../utils/policy-risk';
|
||||
@@ -1146,6 +1147,7 @@ export class DatabaseService {
|
||||
this.migrateEncryptNodeTokens();
|
||||
this.migrateSSOColumns();
|
||||
this.migrateRegistries();
|
||||
this.migrateRegistryDelivery();
|
||||
this.migrateRoleAssignments();
|
||||
this.migrateNotificationRoutes();
|
||||
this.migrateNotificationRoutesNodeId();
|
||||
@@ -2335,6 +2337,62 @@ stmt.run('gitops_schema_version', '1');
|
||||
`);
|
||||
}
|
||||
|
||||
private migrateRegistryDelivery(): void {
|
||||
this.db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS registry_delivery_events (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
event_id TEXT UNIQUE NOT NULL,
|
||||
delivery_source_id TEXT NOT NULL,
|
||||
stack TEXT,
|
||||
op TEXT,
|
||||
attestation_jti TEXT,
|
||||
prep_id_sha256 TEXT,
|
||||
temp_dir_id TEXT,
|
||||
event_type TEXT NOT NULL,
|
||||
source_hash TEXT,
|
||||
pruned_through_seq INTEGER,
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_registry_delivery_events_source_seq
|
||||
ON registry_delivery_events(delivery_source_id, seq);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS registry_delivery_imported_events (
|
||||
event_id TEXT NOT NULL,
|
||||
delivery_source_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
hub_node_id_snapshot INTEGER NOT NULL,
|
||||
stack TEXT,
|
||||
op TEXT,
|
||||
attestation_jti TEXT,
|
||||
prep_id_sha256 TEXT,
|
||||
temp_dir_id TEXT,
|
||||
event_type TEXT NOT NULL,
|
||||
source_hash TEXT,
|
||||
pruned_through_seq INTEGER,
|
||||
created_at INTEGER NOT NULL,
|
||||
imported_at INTEGER NOT NULL,
|
||||
PRIMARY KEY (delivery_source_id, event_id)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS registry_delivery_import_cursor (
|
||||
delivery_source_id TEXT PRIMARY KEY,
|
||||
last_seq INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
`);
|
||||
|
||||
const existing = this.db.prepare(
|
||||
'SELECT value FROM global_settings WHERE key = ?',
|
||||
).get('delivery_source_id') as { value: string } | undefined;
|
||||
if (!existing) {
|
||||
const id = crypto.randomUUID();
|
||||
this.db.prepare(
|
||||
'INSERT INTO global_settings (key, value) VALUES (?, ?)',
|
||||
).run('delivery_source_id', id);
|
||||
this.cachedGlobalSettings = null;
|
||||
}
|
||||
}
|
||||
|
||||
private migrateRoleAssignments(): void {
|
||||
this.db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS role_assignments (
|
||||
@@ -6072,6 +6130,149 @@ stmt.run('gitops_schema_version', '1');
|
||||
this.db.prepare('DELETE FROM audit_log WHERE timestamp < ?').run(cutoff);
|
||||
}
|
||||
|
||||
public insertRegistryDeliveryEvent(params: {
|
||||
deliverySourceId: string;
|
||||
eventType: string;
|
||||
stack?: string | null;
|
||||
op?: string | null;
|
||||
attestationJti?: string | null;
|
||||
prepIdSha256?: string | null;
|
||||
tempDirId?: string | null;
|
||||
sourceHash?: string | null;
|
||||
prunedThroughSeq?: number | null;
|
||||
}): number {
|
||||
const eventId = crypto.randomUUID();
|
||||
const createdAt = Date.now();
|
||||
const result = this.db.prepare(`
|
||||
INSERT INTO registry_delivery_events (
|
||||
event_id, delivery_source_id, stack, op, attestation_jti,
|
||||
prep_id_sha256, temp_dir_id, event_type, source_hash, pruned_through_seq, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`).run(
|
||||
eventId,
|
||||
params.deliverySourceId,
|
||||
params.stack ?? null,
|
||||
params.op ?? null,
|
||||
params.attestationJti ?? null,
|
||||
params.prepIdSha256 ?? null,
|
||||
params.tempDirId ?? null,
|
||||
params.eventType,
|
||||
params.sourceHash ?? null,
|
||||
params.prunedThroughSeq ?? null,
|
||||
createdAt,
|
||||
);
|
||||
return Number(result.lastInsertRowid);
|
||||
}
|
||||
|
||||
public listRegistryDeliveryEvents(
|
||||
deliverySourceId: string,
|
||||
afterSeq: number,
|
||||
limit: number,
|
||||
): import('../types/registryDeliveryEvidence').RegistryDeliveryEventRow[] {
|
||||
return this.db.prepare(`
|
||||
SELECT seq, event_id, delivery_source_id, stack, op, attestation_jti,
|
||||
prep_id_sha256, temp_dir_id, event_type, source_hash, pruned_through_seq, created_at
|
||||
FROM registry_delivery_events
|
||||
WHERE delivery_source_id = ? AND seq > ?
|
||||
ORDER BY seq ASC
|
||||
LIMIT ?
|
||||
`).all(deliverySourceId, afterSeq, limit) as import('../types/registryDeliveryEvidence').RegistryDeliveryEventRow[];
|
||||
}
|
||||
|
||||
public getRegistryDeliveryImportCursor(deliverySourceId: string): number {
|
||||
const row = this.db.prepare(
|
||||
'SELECT last_seq FROM registry_delivery_import_cursor WHERE delivery_source_id = ?',
|
||||
).get(deliverySourceId) as { last_seq: number } | undefined;
|
||||
return row?.last_seq ?? 0;
|
||||
}
|
||||
|
||||
public importRegistryDeliveryEventPage(
|
||||
hubNodeIdSnapshot: number,
|
||||
deliverySourceId: string,
|
||||
events: import('../types/registryDeliveryEvidence').RegistryDeliveryEventRow[],
|
||||
): { imported: number; lastSeq: number } {
|
||||
if (events.length === 0) {
|
||||
return {
|
||||
imported: 0,
|
||||
lastSeq: this.getRegistryDeliveryImportCursor(deliverySourceId),
|
||||
};
|
||||
}
|
||||
|
||||
const importPage = this.db.transaction((rows: import('../types/registryDeliveryEvidence').RegistryDeliveryEventRow[]) => {
|
||||
const insert = this.db.prepare(`
|
||||
INSERT OR IGNORE INTO registry_delivery_imported_events (
|
||||
event_id, delivery_source_id, seq, hub_node_id_snapshot,
|
||||
stack, op, attestation_jti, prep_id_sha256, temp_dir_id,
|
||||
event_type, source_hash, pruned_through_seq, created_at, imported_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`);
|
||||
let imported = 0;
|
||||
let maxSeq = this.getRegistryDeliveryImportCursor(deliverySourceId);
|
||||
const now = Date.now();
|
||||
for (const row of rows) {
|
||||
const result = insert.run(
|
||||
row.event_id,
|
||||
deliverySourceId,
|
||||
row.seq,
|
||||
hubNodeIdSnapshot,
|
||||
row.stack,
|
||||
row.op,
|
||||
row.attestation_jti,
|
||||
row.prep_id_sha256,
|
||||
row.temp_dir_id,
|
||||
row.event_type,
|
||||
row.source_hash,
|
||||
row.pruned_through_seq,
|
||||
row.created_at,
|
||||
now,
|
||||
);
|
||||
if (result.changes > 0) imported += 1;
|
||||
if (row.seq > maxSeq) maxSeq = row.seq;
|
||||
}
|
||||
this.db.prepare(`
|
||||
INSERT INTO registry_delivery_import_cursor (delivery_source_id, last_seq, updated_at)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(delivery_source_id) DO UPDATE SET
|
||||
last_seq = CASE
|
||||
WHEN excluded.last_seq > registry_delivery_import_cursor.last_seq
|
||||
THEN excluded.last_seq
|
||||
ELSE registry_delivery_import_cursor.last_seq
|
||||
END,
|
||||
updated_at = excluded.updated_at
|
||||
`).run(deliverySourceId, maxSeq, now);
|
||||
return { imported, lastSeq: maxSeq };
|
||||
});
|
||||
|
||||
return importPage(events);
|
||||
}
|
||||
|
||||
public cleanupOldDeliveryEvents(daysToKeep = 90): number {
|
||||
const deliverySourceId = this.getGlobalSettings().delivery_source_id;
|
||||
if (!deliverySourceId) return 0;
|
||||
|
||||
const cutoff = Date.now() - (daysToKeep * 24 * 60 * 60 * 1000);
|
||||
const maxPruned = this.db.prepare(`
|
||||
SELECT MAX(seq) as maxSeq FROM registry_delivery_events
|
||||
WHERE delivery_source_id = ? AND created_at < ?
|
||||
`).get(deliverySourceId, cutoff) as { maxSeq: number | null } | undefined;
|
||||
const prunedThrough = maxPruned?.maxSeq ?? null;
|
||||
if (prunedThrough === null) return 0;
|
||||
|
||||
const result = this.db.prepare(`
|
||||
DELETE FROM registry_delivery_events
|
||||
WHERE delivery_source_id = ? AND created_at < ?
|
||||
`).run(deliverySourceId, cutoff);
|
||||
|
||||
if (result.changes > 0) {
|
||||
this.insertRegistryDeliveryEvent({
|
||||
deliverySourceId,
|
||||
eventType: 'retention_gap',
|
||||
prunedThroughSeq: prunedThrough,
|
||||
});
|
||||
}
|
||||
return result.changes;
|
||||
}
|
||||
|
||||
public getAuditLogsInRange(from: number, to: number, limit?: number): AuditLogEntry[] {
|
||||
this.flushAuditLogBuffer();
|
||||
if (limit !== undefined) {
|
||||
|
||||
@@ -1486,6 +1486,37 @@ export class FileSystemService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy managed backup-slot files into destDir for registry-delivery preparation.
|
||||
* Skips integrity markers; only regular compose and env files are copied.
|
||||
*/
|
||||
async copyBackupSlotToDir(stackName: string, destDir: string): Promise<void> {
|
||||
const backupRoot = path.resolve(getBackupBaseDir());
|
||||
const backupDir = path.resolve(backupRoot, String(this.nodeId), stackName);
|
||||
if (!backupDir.startsWith(backupRoot + path.sep)) {
|
||||
throw Object.assign(new Error('Path escapes backup directory'), { code: 'INVALID_PATH' });
|
||||
}
|
||||
await fsPromises.mkdir(destDir, { recursive: true, mode: 0o700 });
|
||||
const items = await fsPromises.readdir(backupDir);
|
||||
for (const item of items) {
|
||||
if (item === '.checksums' || item === '.timestamp') continue;
|
||||
const src = path.resolve(backupDir, item);
|
||||
if (!src.startsWith(backupDir + path.sep)) continue;
|
||||
const stat = await fsPromises.lstat(src);
|
||||
if (!stat.isFile() || stat.isSymbolicLink()) continue;
|
||||
const dest = path.join(destDir, item);
|
||||
await fsPromises.copyFile(src, dest, fsPromises.constants.COPYFILE_EXCL).catch(async (err: NodeJS.ErrnoException) => {
|
||||
if (err.code === 'EEXIST') {
|
||||
await fsPromises.unlink(dest);
|
||||
await fsPromises.copyFile(src, dest);
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
});
|
||||
await fsPromises.chmod(dest, 0o600);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Stack-scoped file explorer methods
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import { promises as fsPromises, existsSync } from 'fs';
|
||||
import { spawn } from 'child_process';
|
||||
import crypto from 'crypto';
|
||||
import os from 'os';
|
||||
import path from 'path';
|
||||
@@ -43,9 +42,12 @@ import {
|
||||
stackManagedRoot,
|
||||
} from './gitops/directApplication';
|
||||
import type { GitOpsApplicationRow } from './gitops/types';
|
||||
import { appliedRelPathFor, candidateRelPathForSha, deleteStagingMarker, readStagingMarker, writeStagingMarker } from './gitops/createStagingMarker';
|
||||
import { appliedRelPathFor, candidateRelPathForSha, deleteStagingMarker, readStagingMarker, validateCandidateRelPath, writeStagingMarker } from './gitops/createStagingMarker';
|
||||
import { cleanupUnclaimedManagedRoot, removeOperationOwnedPaths } from './gitops/createCleanup';
|
||||
import { managedAreaBase } from './gitops/managedPaths';
|
||||
import { getRegistryDeliveryContext, getRegistryDeliveryLockContext } from '../helpers/registryDeliveryContext';
|
||||
import { copyPreparedPayloadDirectory } from '../helpers/registryDeliveryMaterialize';
|
||||
import { runDockerCompose as spawnDockerCompose } from '../helpers/dockerComposeRunner';
|
||||
|
||||
/**
|
||||
* GitSourceService - fetch compose files from a Git repository and apply
|
||||
@@ -1222,9 +1224,9 @@ export class GitSourceService {
|
||||
const startedAt = Date.now();
|
||||
const diag = isDebugEnabled();
|
||||
if (diag) {
|
||||
console.log(
|
||||
`[GitSource:diag] fetch start host=${sanitizeForLog(repoHost(repoUrl))} branch=${sanitizeForLog(branch)} files=${composePaths.length} envSync=${envPath ? 'true' : 'false'}`
|
||||
);
|
||||
console.log(sanitizeForLog(
|
||||
`[GitSource:diag] fetch start host=${repoHost(repoUrl)} branch=${branch} files=${composePaths.length} envSync=${envPath ? 'true' : 'false'}`,
|
||||
));
|
||||
}
|
||||
|
||||
try {
|
||||
@@ -1544,26 +1546,8 @@ export class GitSourceService {
|
||||
};
|
||||
}
|
||||
|
||||
private runDockerCompose(args: string[], cwd: string, timeoutMs: number): Promise<{ code: number; stdout: string; stderr: string }> {
|
||||
return new Promise((resolve) => {
|
||||
const child = spawn('docker', args, { cwd });
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
const timer = setTimeout(() => {
|
||||
try { child.kill('SIGKILL'); } catch { /* best effort */ }
|
||||
resolve({ code: -1, stdout, stderr: stderr + '\nValidation timed out.' });
|
||||
}, timeoutMs);
|
||||
child.stdout.on('data', d => { stdout += d.toString(); });
|
||||
child.stderr.on('data', d => { stderr += d.toString(); });
|
||||
child.on('close', (code) => {
|
||||
clearTimeout(timer);
|
||||
resolve({ code: code ?? -1, stdout, stderr });
|
||||
});
|
||||
child.on('error', (err) => {
|
||||
clearTimeout(timer);
|
||||
resolve({ code: -1, stdout, stderr: stderr + '\n' + err.message });
|
||||
});
|
||||
});
|
||||
private runDockerCompose(args: string[], cwd: string, timeoutMs: number) {
|
||||
return spawnDockerCompose(args, cwd, timeoutMs);
|
||||
}
|
||||
|
||||
// ─── Hashing + diff ──────────────────────────────────────────────────────
|
||||
@@ -2205,6 +2189,7 @@ export class GitSourceService {
|
||||
'git_apply',
|
||||
opts.actor ?? 'system:git-source',
|
||||
() => this.applyLocked(stackName, commitSha, opts),
|
||||
getRegistryDeliveryLockContext(),
|
||||
);
|
||||
if (!lock.ran) {
|
||||
throw new GitSourceError(
|
||||
@@ -2348,6 +2333,15 @@ export class GitSourceService {
|
||||
|
||||
// The staged candidate must still exist and be complete; a deleted
|
||||
// candidate (or a node restart that swept it) invalidates the pull.
|
||||
const deliveryPrepId = getRegistryDeliveryContext()?.envelope.prepId;
|
||||
if (deliveryPrepId) {
|
||||
await this.restoreApplyFromPreparedGitCandidate(
|
||||
deliveryPrepId,
|
||||
stackName,
|
||||
commitSha,
|
||||
pending.candidateRelPath,
|
||||
);
|
||||
}
|
||||
const dataDir = process.env.DATA_DIR || path.join(process.cwd(), 'data');
|
||||
const candidateAbs = path.join(dataDir, 'git-managed', String(nodeId), stackName, pending.candidateRelPath);
|
||||
try {
|
||||
@@ -2801,6 +2795,7 @@ export class GitSourceService {
|
||||
// the complete-project candidate inside the clone lifecycle.
|
||||
const manifestSvc = GitProjectManifestService.getInstance();
|
||||
const materialization: { value: MaterializationResult | null } = { value: null };
|
||||
const deliveryPrepId = getRegistryDeliveryContext()?.envelope.prepId;
|
||||
let fetched: FetchResult;
|
||||
const createFetchAuth = input.authType === 'token'
|
||||
? { token: input.token }
|
||||
@@ -2813,31 +2808,43 @@ export class GitSourceService {
|
||||
}
|
||||
: { token: null };
|
||||
try {
|
||||
fetched = await this.fetchFromGit({
|
||||
repoUrl: input.repoUrl,
|
||||
branch: input.branch,
|
||||
composePaths: input.composePaths,
|
||||
envPath: input.syncEnv ? input.envPath : null,
|
||||
...createFetchAuth,
|
||||
onClone: async (cloneDir, commitSha, envContent) => {
|
||||
// The candidate path is recorded before the build that
|
||||
// creates it, so a crash mid-build still names exactly one
|
||||
// directory this operation owns.
|
||||
staged.candidateRelPath = candidateRelPathForSha(commitSha);
|
||||
await writeStagingMarker(managedRoot, {
|
||||
schemaVersion: 1,
|
||||
operationId: gitopsOperationId,
|
||||
rootPreexisted,
|
||||
candidateRelPath: staged.candidateRelPath,
|
||||
createdAt: Date.now(),
|
||||
});
|
||||
materialization.value = await this.buildMaterialization(input.stackName, cloneDir, commitSha, {
|
||||
compose_paths: input.composePaths,
|
||||
context_dir: input.contextDir,
|
||||
sync_env: input.syncEnv,
|
||||
}, envContent);
|
||||
},
|
||||
});
|
||||
if (deliveryPrepId) {
|
||||
const restored = await this.restoreCreateFromPreparedGitCandidate(
|
||||
deliveryPrepId,
|
||||
managedRoot,
|
||||
rootPreexisted,
|
||||
gitopsOperationId,
|
||||
staged,
|
||||
);
|
||||
fetched = restored.fetched;
|
||||
materialization.value = restored.materialization;
|
||||
} else {
|
||||
fetched = await this.fetchFromGit({
|
||||
repoUrl: input.repoUrl,
|
||||
branch: input.branch,
|
||||
composePaths: input.composePaths,
|
||||
envPath: input.syncEnv ? input.envPath : null,
|
||||
...createFetchAuth,
|
||||
onClone: async (cloneDir, commitSha, envContent) => {
|
||||
// The candidate path is recorded before the build that
|
||||
// creates it, so a crash mid-build still names exactly one
|
||||
// directory this operation owns.
|
||||
staged.candidateRelPath = candidateRelPathForSha(commitSha);
|
||||
await writeStagingMarker(managedRoot, {
|
||||
schemaVersion: 1,
|
||||
operationId: gitopsOperationId,
|
||||
rootPreexisted,
|
||||
candidateRelPath: staged.candidateRelPath,
|
||||
createdAt: Date.now(),
|
||||
});
|
||||
materialization.value = await this.buildMaterialization(input.stackName, cloneDir, commitSha, {
|
||||
compose_paths: input.composePaths,
|
||||
context_dir: input.contextDir,
|
||||
sync_env: input.syncEnv,
|
||||
}, envContent);
|
||||
},
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
// Materialization refuses routinely, not just on crashes. The
|
||||
// marker has to come off with the staged files, or it would
|
||||
@@ -3684,6 +3691,201 @@ export class GitSourceService {
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Registry delivery preparation ─────────────────────────────────────
|
||||
|
||||
private async loadPreparedGitCandidate(
|
||||
prepId: string,
|
||||
managedRoot: string,
|
||||
expectations?: { commitSha?: string; candidateRelPath?: string },
|
||||
): Promise<import('../helpers/registryDeliveryGitCandidate').GitCandidatePreparedMeta> {
|
||||
const { PreparedSourceStore } = await import('./preparedSourceStore');
|
||||
const {
|
||||
installGitCandidatePayloadToManagedRoot,
|
||||
readGitCandidatePreparedMeta,
|
||||
} = await import('../helpers/registryDeliveryGitCandidate');
|
||||
const payloadPath = PreparedSourceStore.getInstance().peekPayloadPath(prepId);
|
||||
const meta = await readGitCandidatePreparedMeta(payloadPath);
|
||||
if (!meta.materialization.validation.ok) {
|
||||
throw new GitSourceError(
|
||||
'GIT_ERROR',
|
||||
`Compose validation failed: ${meta.materialization.validation.error ?? 'unknown'}`,
|
||||
);
|
||||
}
|
||||
if (expectations?.commitSha && meta.commitSha !== expectations.commitSha) {
|
||||
throw new GitSourceError('GIT_ERROR', 'Prepared git candidate commit mismatch');
|
||||
}
|
||||
if (expectations?.candidateRelPath && meta.candidateRelPath !== expectations.candidateRelPath) {
|
||||
throw new GitSourceError('GIT_ERROR', 'Prepared git candidate path mismatch');
|
||||
}
|
||||
await installGitCandidatePayloadToManagedRoot(payloadPath, managedRoot, meta.candidateRelPath);
|
||||
return meta;
|
||||
}
|
||||
|
||||
private async restoreCreateFromPreparedGitCandidate(
|
||||
prepId: string,
|
||||
managedRoot: string,
|
||||
rootPreexisted: boolean,
|
||||
gitopsOperationId: string,
|
||||
staged: { candidateRelPath: string | null },
|
||||
): Promise<{ fetched: FetchResult; materialization: MaterializationResult }> {
|
||||
const { fetchResultFromPreparedMeta } = await import('../helpers/registryDeliveryGitCandidate');
|
||||
const meta = await this.loadPreparedGitCandidate(prepId, managedRoot);
|
||||
staged.candidateRelPath = meta.candidateRelPath;
|
||||
await writeStagingMarker(managedRoot, {
|
||||
schemaVersion: 1,
|
||||
operationId: gitopsOperationId,
|
||||
rootPreexisted,
|
||||
candidateRelPath: staged.candidateRelPath,
|
||||
createdAt: Date.now(),
|
||||
});
|
||||
return {
|
||||
fetched: fetchResultFromPreparedMeta(meta),
|
||||
materialization: meta.materialization,
|
||||
};
|
||||
}
|
||||
|
||||
private async restoreApplyFromPreparedGitCandidate(
|
||||
prepId: string,
|
||||
stackName: string,
|
||||
commitSha: string,
|
||||
candidateRelPath: string,
|
||||
): Promise<void> {
|
||||
const managedRoot = path.resolve(stackManagedRoot(stackName));
|
||||
await this.loadPreparedGitCandidate(prepId, managedRoot, { commitSha, candidateRelPath });
|
||||
}
|
||||
|
||||
public async prepareRegistryDeliveryFromGit(
|
||||
input: CreateStackFromGitInput,
|
||||
): Promise<{ prepId: string; sourceHash: string }> {
|
||||
const materialization: { value: MaterializationResult | null } = { value: null };
|
||||
const fetched = await this.fetchFromGit({
|
||||
repoUrl: input.repoUrl,
|
||||
branch: input.branch,
|
||||
composePaths: input.composePaths,
|
||||
envPath: input.syncEnv ? input.envPath : null,
|
||||
token: input.token,
|
||||
onClone: async (cloneDir, commitSha, envContent) => {
|
||||
materialization.value = await this.buildMaterialization(
|
||||
input.stackName,
|
||||
cloneDir,
|
||||
commitSha,
|
||||
{
|
||||
compose_paths: input.composePaths,
|
||||
context_dir: input.contextDir,
|
||||
sync_env: input.syncEnv,
|
||||
},
|
||||
envContent,
|
||||
);
|
||||
},
|
||||
});
|
||||
if (!materialization.value?.validation.ok) {
|
||||
throw new GitSourceError(
|
||||
'GIT_ERROR',
|
||||
`Compose validation failed: ${materialization.value?.validation.error ?? 'unknown'}`,
|
||||
);
|
||||
}
|
||||
const managedRoot = path.resolve(stackManagedRoot(input.stackName));
|
||||
const candidateRel = materialization.value.candidateRelPath;
|
||||
const pathReason = validateCandidateRelPath(candidateRel, managedRoot);
|
||||
if (pathReason) {
|
||||
throw new GitSourceError('GIT_ERROR', pathReason);
|
||||
}
|
||||
const candidateAbs = path.resolve(managedRoot, candidateRel);
|
||||
if (!candidateAbs.startsWith(managedRoot + path.sep)) {
|
||||
throw new GitSourceError('GIT_ERROR', 'Invalid candidate path');
|
||||
}
|
||||
const stagingDir = await fsPromises.mkdtemp(path.join(os.tmpdir(), 'sencho-regprep-'));
|
||||
try {
|
||||
await copyPreparedPayloadDirectory(candidateAbs, stagingDir);
|
||||
const { writeGitCandidatePreparedMeta } = await import('../helpers/registryDeliveryGitCandidate');
|
||||
await writeGitCandidatePreparedMeta(stagingDir, {
|
||||
version: 1,
|
||||
commitSha: fetched.commitSha,
|
||||
resolvedRefKind: fetched.resolvedRefKind,
|
||||
candidateRelPath: materialization.value.candidateRelPath,
|
||||
composeFiles: fetched.composeFiles,
|
||||
envContent: fetched.envContent,
|
||||
materialization: materialization.value,
|
||||
warnings: fetched.warnings,
|
||||
});
|
||||
const { hashDeliverySourceDir } = await import('../helpers/registryDeliveryHashes');
|
||||
const { PreparedSourceStore } = await import('./preparedSourceStore');
|
||||
const sourceHash = hashDeliverySourceDir(stagingDir);
|
||||
const entry = await PreparedSourceStore.getInstance().prepareFromDirectory(
|
||||
'git-candidate',
|
||||
sourceHash,
|
||||
stagingDir,
|
||||
);
|
||||
return { prepId: entry.prepId, sourceHash };
|
||||
} catch (error) {
|
||||
await fsPromises.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
throw error;
|
||||
} finally {
|
||||
const rmTarget = path.resolve(candidateAbs);
|
||||
if (rmTarget.startsWith(managedRoot + path.sep)) {
|
||||
// Canonical js/path-injection barrier inline with the rm sink.
|
||||
await fsPromises.rm(rmTarget, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public async prepareRegistryDeliveryFromPending(
|
||||
stackName: string,
|
||||
): Promise<{ prepId: string; sourceHash: string }> {
|
||||
const src = DatabaseService.getInstance().getGitSource(stackName);
|
||||
if (!src?.pending_commit_sha || !src.pending_compose_content) {
|
||||
throw new GitSourceError('GIT_ERROR', 'No pending pull to prepare');
|
||||
}
|
||||
const pending = this.decodePendingCompose(src.pending_compose_content);
|
||||
if (!pending.candidateRelPath || pending.inventory === null) {
|
||||
throw new GitSourceError('GIT_ERROR', 'No staged candidate for pending apply');
|
||||
}
|
||||
const envContent = src.pending_env_content !== null
|
||||
? this.crypto.decrypt(src.pending_env_content)
|
||||
: null;
|
||||
const managedRoot = path.resolve(stackManagedRoot(stackName));
|
||||
const pathReason = validateCandidateRelPath(pending.candidateRelPath, managedRoot);
|
||||
if (pathReason) {
|
||||
throw new GitSourceError('GIT_ERROR', pathReason);
|
||||
}
|
||||
const candidateAbs = path.resolve(managedRoot, pending.candidateRelPath);
|
||||
if (!candidateAbs.startsWith(managedRoot + path.sep)) {
|
||||
throw new GitSourceError('GIT_ERROR', 'Invalid candidate path');
|
||||
}
|
||||
const stagingDir = await fsPromises.mkdtemp(path.join(os.tmpdir(), 'sencho-regprep-'));
|
||||
try {
|
||||
await copyPreparedPayloadDirectory(candidateAbs, stagingDir);
|
||||
const { writeGitCandidatePreparedMeta } = await import('../helpers/registryDeliveryGitCandidate');
|
||||
await writeGitCandidatePreparedMeta(stagingDir, {
|
||||
version: 1,
|
||||
commitSha: src.pending_commit_sha,
|
||||
resolvedRefKind: priorFetchIdentity(this.gitopsApplicationFor(stackName))?.kind ?? 'branch',
|
||||
candidateRelPath: pending.candidateRelPath,
|
||||
composeFiles: pending.files,
|
||||
envContent,
|
||||
materialization: {
|
||||
inventory: pending.inventory,
|
||||
contextCopyPlans: [],
|
||||
candidateRelPath: pending.candidateRelPath,
|
||||
validation: { ok: true },
|
||||
},
|
||||
warnings: [],
|
||||
});
|
||||
const { hashDeliverySourceDir } = await import('../helpers/registryDeliveryHashes');
|
||||
const { PreparedSourceStore } = await import('./preparedSourceStore');
|
||||
const sourceHash = hashDeliverySourceDir(stagingDir);
|
||||
const entry = await PreparedSourceStore.getInstance().prepareFromDirectory(
|
||||
'git-candidate',
|
||||
sourceHash,
|
||||
stagingDir,
|
||||
);
|
||||
return { prepId: entry.prepId, sourceHash };
|
||||
} catch (error) {
|
||||
await fsPromises.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Concurrency ─────────────────────────────────────────────────────────
|
||||
|
||||
private async withStackLock<T>(stackName: string, fn: () => Promise<T>): Promise<T> {
|
||||
|
||||
@@ -24,6 +24,7 @@ import { isPathWithinBase, isValidStackName, isValidRelativeStackPath } from '..
|
||||
import { getErrorMessage } from '../utils/errors';
|
||||
import { PORT as SENCHO_LISTEN_PORT } from '../helpers/constants';
|
||||
import { assertPolicyGateAllows, buildSystemPolicyGateOptions } from '../helpers/policyGate';
|
||||
import { prepareOutboundRegistryDeliveryBody } from '../helpers/registryDeliveryOutbound';
|
||||
|
||||
const ACTIVITY_BUFFER_SIZE = 1000;
|
||||
const ALIAS_REFRESH_INTERVAL_MS = 60_000;
|
||||
@@ -2497,10 +2498,30 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
if (target.apiToken) headers['Authorization'] = `Bearer ${target.apiToken}`;
|
||||
const proxyHeaders = LicenseService.getInstance().getProxyHeaders();
|
||||
headers[PROXY_TIER_HEADER] = proxyHeaders.tier;
|
||||
|
||||
let bodyToSend: unknown = body;
|
||||
if (method === 'POST' || method === 'PUT') {
|
||||
const bodyRecord = body === undefined || body === null
|
||||
? {}
|
||||
: (typeof body === 'object' && !Array.isArray(body) ? body as Record<string, unknown> : null);
|
||||
if (bodyRecord !== null) {
|
||||
const augmented = await prepareOutboundRegistryDeliveryBody({
|
||||
method,
|
||||
apiPath,
|
||||
nodeId,
|
||||
body: bodyRecord,
|
||||
});
|
||||
if (!augmented.ok) {
|
||||
throw new MeshError('push_failed', augmented.error);
|
||||
}
|
||||
bodyToSend = augmented.body;
|
||||
}
|
||||
}
|
||||
|
||||
return await fetch(url, {
|
||||
method,
|
||||
headers,
|
||||
body: body === undefined ? undefined : JSON.stringify(body),
|
||||
body: bodyToSend === undefined ? undefined : JSON.stringify(bodyToSend),
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -706,6 +706,7 @@ export class MonitorService {
|
||||
const notifSummary = db.cleanupOldNotifications(isNaN(retentionDays) ? 30 : retentionDays);
|
||||
const auditRetentionDays = parseInt(settings['audit_retention_days'] || '90', 10);
|
||||
db.cleanupOldAuditLogs(isNaN(auditRetentionDays) ? 90 : auditRetentionDays);
|
||||
db.cleanupOldDeliveryEvents(isNaN(auditRetentionDays) ? 90 : auditRetentionDays);
|
||||
const scanPerImage = parseInt(settings['scan_history_per_image_limit'] || '50', 10);
|
||||
const scanPruned = db.pruneScanHistoryPerImage(isNaN(scanPerImage) ? 50 : scanPerImage);
|
||||
if (isDebugEnabled()) console.log(`[Monitor:diag] Cleanup: metrics ${isNaN(retentionHours) ? 24 : retentionHours}h, notifications ${isNaN(retentionDays) ? 30 : retentionDays}d (ttl=${notifSummary.ttl} perStack=${notifSummary.perStack} perNode=${notifSummary.perNode}), audit ${isNaN(auditRetentionDays) ? 90 : auditRetentionDays}d, scans pruned ${scanPruned}`);
|
||||
|
||||
@@ -368,16 +368,28 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
|
||||
if (s) this.teardownStream(s);
|
||||
this.removeStream(streamId);
|
||||
});
|
||||
req.on('aborted', () => {
|
||||
if (this.streams.has(streamId)) {
|
||||
this.notifyHttpClientAbort(streamId);
|
||||
this.removeStream(streamId);
|
||||
}
|
||||
});
|
||||
|
||||
res.on('close', () => {
|
||||
// Client disconnected before response finished.
|
||||
if (this.streams.has(streamId)) {
|
||||
this.notifyHttpClientAbort(streamId);
|
||||
this.removeStream(streamId);
|
||||
this.sendJson({ t: 'http_err', s: streamId, code: 'tunnel_down', message: 'client aborted' });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Notify the agent that the loopback HTTP client disconnected mid-request. */
|
||||
private notifyHttpClientAbort(streamId: number): void {
|
||||
this.sendJson({ t: 'http_cancel', s: streamId });
|
||||
this.sendJson({ t: 'http_err', s: streamId, code: 'tunnel_down', message: 'client aborted' });
|
||||
}
|
||||
|
||||
private handleLoopbackUpgrade(req: IncomingMessage, socket: Socket, head: Buffer): void {
|
||||
if (this.closed || this.tunnelWs.readyState !== WebSocket.OPEN) {
|
||||
socket.write('HTTP/1.1 502 Bad Gateway\r\n\r\n');
|
||||
|
||||
@@ -79,6 +79,7 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
private static instance: PilotTunnelManager;
|
||||
private bridges: Map<number, PilotTunnelBridge> = new Map();
|
||||
private bridgeKinds: Map<number, BridgeKind> = new Map();
|
||||
private tunnelConfidential: Map<number, boolean> = new Map();
|
||||
private softWarned = false;
|
||||
|
||||
private constructor() {
|
||||
@@ -104,6 +105,7 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
}
|
||||
PilotTunnelManager.instance.bridges.clear();
|
||||
PilotTunnelManager.instance.bridgeKinds.clear();
|
||||
PilotTunnelManager.instance.tunnelConfidential.clear();
|
||||
}
|
||||
PilotTunnelManager.instance = undefined as unknown as PilotTunnelManager;
|
||||
}
|
||||
@@ -125,13 +127,19 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
*
|
||||
* Resolves once the loopback HTTP server is listening.
|
||||
*/
|
||||
public async registerTunnel(nodeId: number, ws: WebSocket, agentVersion?: string): Promise<void> {
|
||||
public async registerTunnel(
|
||||
nodeId: number,
|
||||
ws: WebSocket,
|
||||
agentVersion?: string,
|
||||
tunnelConfidential = false,
|
||||
): Promise<void> {
|
||||
const existing = this.bridges.get(nodeId);
|
||||
const replaced = existing != null;
|
||||
if (existing) {
|
||||
existing.close(PilotCloseCode.Replaced, 'replaced by newer tunnel');
|
||||
this.bridges.delete(nodeId);
|
||||
this.bridgeKinds.delete(nodeId);
|
||||
this.tunnelConfidential.delete(nodeId);
|
||||
}
|
||||
|
||||
// Hard cap: only counts tunnels for *other* nodes since we just
|
||||
@@ -158,6 +166,7 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
if (this.bridges.get(nodeId) === bridge) {
|
||||
this.bridges.delete(nodeId);
|
||||
this.bridgeKinds.delete(nodeId);
|
||||
this.tunnelConfidential.delete(nodeId);
|
||||
DatabaseService.getInstance().updateNodeStatus(nodeId, 'offline');
|
||||
this.emit('tunnel-down', nodeId);
|
||||
}
|
||||
@@ -166,6 +175,7 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
|
||||
this.bridges.set(nodeId, bridge);
|
||||
this.bridgeKinds.set(nodeId, 'pilot');
|
||||
this.tunnelConfidential.set(nodeId, tunnelConfidential);
|
||||
const db = DatabaseService.getInstance();
|
||||
db.updateNodeStatus(nodeId, 'online');
|
||||
db.updateNode(nodeId, {
|
||||
@@ -179,6 +189,11 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
this.emit('tunnel-up', nodeId);
|
||||
}
|
||||
|
||||
/** Whether the active pilot tunnel was established over confidential transport. */
|
||||
public isTunnelConfidential(nodeId: number): boolean {
|
||||
return this.tunnelConfidential.get(nodeId) === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-tunnel breakdown for the metrics endpoint. Includes the
|
||||
* loopback-relative connectedAt and bufferedAmount so one-bad-node cases
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
import axios from 'axios';
|
||||
|
||||
import {
|
||||
importRegistryDeliveryEvidencePage,
|
||||
} from '../helpers/registryDeliveryEvidence';
|
||||
import type { RegistryDeliveryEvidencePage } from '../types/registryDeliveryEvidence';
|
||||
import { getErrorMessage } from '../utils/errors';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
import { NodeRegistry } from './NodeRegistry';
|
||||
import { PilotTunnelManager } from './PilotTunnelManager';
|
||||
|
||||
const RECONCILE_INTERVAL_MS = 5 * 60 * 1000;
|
||||
const RECONCILE_INITIAL_DELAY_MS = 30_000;
|
||||
const EVIDENCE_PAGE_LIMIT = 100;
|
||||
const NODE_FAILURE_BACKOFF_MS = 15 * 60 * 1000;
|
||||
|
||||
interface NodeBackoffState {
|
||||
until: number;
|
||||
failures: number;
|
||||
}
|
||||
|
||||
export class RegistryDeliveryReconciler {
|
||||
private static instance: RegistryDeliveryReconciler | null = null;
|
||||
private intervalHandle: ReturnType<typeof setInterval> | null = null;
|
||||
private initialTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private running = false;
|
||||
private stopped = false;
|
||||
private readonly nodeBackoff = new Map<number, NodeBackoffState>();
|
||||
private readonly lastSourceIdByNode = new Map<number, string>();
|
||||
|
||||
static getInstance(): RegistryDeliveryReconciler {
|
||||
if (!this.instance) this.instance = new RegistryDeliveryReconciler();
|
||||
return this.instance;
|
||||
}
|
||||
|
||||
static resetForTests(): void {
|
||||
this.instance?.stop();
|
||||
this.instance = null;
|
||||
}
|
||||
|
||||
private constructor() { /* singleton */ }
|
||||
|
||||
start(): void {
|
||||
if (this.intervalHandle || this.initialTimer) return;
|
||||
this.stopped = false;
|
||||
this.initialTimer = setTimeout(() => {
|
||||
this.initialTimer = null;
|
||||
if (this.stopped) return;
|
||||
void this.tick();
|
||||
this.intervalHandle = setInterval(() => void this.tick(), RECONCILE_INTERVAL_MS);
|
||||
if (typeof this.intervalHandle.unref === 'function') {
|
||||
this.intervalHandle.unref();
|
||||
}
|
||||
}, RECONCILE_INITIAL_DELAY_MS);
|
||||
if (typeof this.initialTimer.unref === 'function') {
|
||||
this.initialTimer.unref();
|
||||
}
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
this.stopped = true;
|
||||
if (this.initialTimer) {
|
||||
clearTimeout(this.initialTimer);
|
||||
this.initialTimer = null;
|
||||
}
|
||||
if (this.intervalHandle) {
|
||||
clearInterval(this.intervalHandle);
|
||||
this.intervalHandle = null;
|
||||
}
|
||||
}
|
||||
|
||||
async tick(): Promise<void> {
|
||||
if (this.running || this.stopped) return;
|
||||
this.running = true;
|
||||
try {
|
||||
const nodes = DatabaseService.getInstance().getNodes()
|
||||
.filter((node) => node.type === 'remote');
|
||||
for (const node of nodes) {
|
||||
if (node.mode === 'pilot_agent' && !PilotTunnelManager.getInstance().hasActiveTunnel(node.id!)) {
|
||||
continue;
|
||||
}
|
||||
await this.reconcileNode(node.id!);
|
||||
}
|
||||
} finally {
|
||||
this.running = false;
|
||||
}
|
||||
}
|
||||
|
||||
async reconcileNode(nodeId: number): Promise<void> {
|
||||
const backoff = this.nodeBackoff.get(nodeId);
|
||||
if (backoff && Date.now() < backoff.until) {
|
||||
return;
|
||||
}
|
||||
|
||||
const target = NodeRegistry.getInstance().getProxyTarget(nodeId);
|
||||
if (!target) {
|
||||
this.markNodeFailure(nodeId);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
let deliverySourceId = this.lastSourceIdByNode.get(nodeId);
|
||||
let cursor = deliverySourceId
|
||||
? DatabaseService.getInstance().getRegistryDeliveryImportCursor(deliverySourceId)
|
||||
: 0;
|
||||
let pages = 0;
|
||||
|
||||
while (pages < 50) {
|
||||
const page = await this.fetchEvidencePage(target, cursor, EVIDENCE_PAGE_LIMIT);
|
||||
deliverySourceId = page.deliverySourceId;
|
||||
this.lastSourceIdByNode.set(nodeId, deliverySourceId);
|
||||
if (page.events.length === 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
const hubNodeId = nodeId;
|
||||
importRegistryDeliveryEvidencePage(hubNodeId, deliverySourceId, page.events);
|
||||
|
||||
cursor = page.nextCursor;
|
||||
pages += 1;
|
||||
if (page.events.length < EVIDENCE_PAGE_LIMIT) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
this.nodeBackoff.delete(nodeId);
|
||||
if (isDebugEnabled() && deliverySourceId) {
|
||||
console.log(
|
||||
`[RegistryDeliveryReconciler:diag] imported evidence from node ${nodeId} source=${deliverySourceId} cursor=${cursor}`,
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
this.markNodeFailure(nodeId);
|
||||
console.warn(
|
||||
`[RegistryDeliveryReconciler] evidence import failed for node ${nodeId}:`,
|
||||
getErrorMessage(error, 'unknown'),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchEvidencePage(
|
||||
target: { apiUrl: string; apiToken: string },
|
||||
cursor: number,
|
||||
limit: number,
|
||||
): Promise<RegistryDeliveryEvidencePage> {
|
||||
const base = target.apiUrl.replace(/\/$/, '');
|
||||
const headers: Record<string, string> = {};
|
||||
if (target.apiToken) {
|
||||
headers.Authorization = `Bearer ${target.apiToken}`;
|
||||
}
|
||||
|
||||
const res = await axios.get(`${base}/api/registry-delivery/evidence`, {
|
||||
headers,
|
||||
params: { cursor, limit },
|
||||
timeout: 30_000,
|
||||
validateStatus: () => true,
|
||||
});
|
||||
|
||||
if (res.status < 200 || res.status >= 300) {
|
||||
const message = typeof res.data?.error === 'string'
|
||||
? res.data.error
|
||||
: 'Registry delivery evidence fetch failed';
|
||||
throw Object.assign(new Error(message), { status: res.status });
|
||||
}
|
||||
|
||||
return res.data as RegistryDeliveryEvidencePage;
|
||||
}
|
||||
|
||||
private markNodeFailure(nodeId: number): void {
|
||||
const existing = this.nodeBackoff.get(nodeId);
|
||||
const failures = (existing?.failures ?? 0) + 1;
|
||||
const backoffMs = Math.min(NODE_FAILURE_BACKOFF_MS * failures, 60 * 60 * 1000);
|
||||
this.nodeBackoff.set(nodeId, {
|
||||
failures,
|
||||
until: Date.now() + backoffMs,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,348 @@
|
||||
import crypto from 'crypto';
|
||||
import jwt from 'jsonwebtoken';
|
||||
import path from 'path';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
import { NodeRegistry } from './NodeRegistry';
|
||||
import {
|
||||
RegistryService,
|
||||
type DockerConfigHostResolution,
|
||||
} from './RegistryService';
|
||||
import { discoverRegistryReferences } from './registryReferenceDiscovery';
|
||||
import { remoteAdvertisesCapability } from '../helpers/remoteCapabilities';
|
||||
import { REMOTE_REGISTRY_CREDENTIALS_CAPABILITY } from './CapabilityRegistry';
|
||||
import { isTrustedProxyPeer } from '../helpers/trustedProxyCidrs';
|
||||
import type { RegistryDeliveryEnvelope, RegistryDeliveryAuthEntry } from '../helpers/registryDeliveryContext';
|
||||
import { classifyRegistryDeliveryOp } from '../helpers/registryOpClassifier';
|
||||
import { prepareSourceForDiscover, resolveBlueprintPostApplyDiscovery } from '../helpers/registryDeliveryPrepare';
|
||||
import { PreparedSourceStore } from './preparedSourceStore';
|
||||
import { hashProjectSource } from '../helpers/registryDeliveryHashes';
|
||||
import { isValidStackName } from '../utils/validation';
|
||||
import {
|
||||
resolveComposeEnvForDiscovery,
|
||||
} from '../helpers/registryDeliveryComposeEnv';
|
||||
|
||||
const ATTESTATION_AUD = 'registry-delivery';
|
||||
const ATTESTATION_TTL_SECONDS = 900;
|
||||
|
||||
export interface RegistryDeliveryDiscoverRequest {
|
||||
stack?: string;
|
||||
op: string;
|
||||
service?: string;
|
||||
sourceKind: string;
|
||||
sourceHash?: string;
|
||||
actionSetHash: string;
|
||||
prepId?: string;
|
||||
envVars?: Record<string, string>;
|
||||
template?: unknown;
|
||||
stackName?: string;
|
||||
git?: Record<string, unknown>;
|
||||
gitApply?: boolean;
|
||||
restoreVariant?: string;
|
||||
composeContent?: string;
|
||||
}
|
||||
|
||||
export interface RegistryDeliveryDiscoverResponse {
|
||||
prepId?: string;
|
||||
referencedHosts: string[];
|
||||
coveredHosts: string[];
|
||||
sourceHash: string;
|
||||
actionSetHash: string;
|
||||
deliverySourceId: string;
|
||||
attestation: string;
|
||||
}
|
||||
|
||||
export class RegistryDeliveryService {
|
||||
private static instance: RegistryDeliveryService | null = null;
|
||||
private readonly targetSessionId = crypto.randomBytes(16).toString('hex');
|
||||
private readonly consumedJtis = new Map<string, number>();
|
||||
private maxConsumedJtis = 10_000;
|
||||
|
||||
static getInstance(): RegistryDeliveryService {
|
||||
if (!this.instance) this.instance = new RegistryDeliveryService();
|
||||
return this.instance;
|
||||
}
|
||||
|
||||
static resetForTests(): void {
|
||||
this.instance = null;
|
||||
}
|
||||
|
||||
/** @internal Narrow replay-store capacity for unit tests. */
|
||||
setReplayStoreCapacityForTests(capacity: number): void {
|
||||
this.maxConsumedJtis = capacity;
|
||||
this.consumedJtis.clear();
|
||||
}
|
||||
|
||||
getTargetSessionId(): string {
|
||||
return this.targetSessionId;
|
||||
}
|
||||
|
||||
getDeliverySourceId(): string {
|
||||
const settings = DatabaseService.getInstance().getGlobalSettings();
|
||||
const id = settings.delivery_source_id;
|
||||
if (!id) {
|
||||
throw new Error('delivery_source_id is not configured');
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
private getJwtSecret(): string {
|
||||
const secret = DatabaseService.getInstance().getGlobalSettings().auth_jwt_secret;
|
||||
if (!secret) {
|
||||
throw new Error('auth_jwt_secret is not configured');
|
||||
}
|
||||
return secret;
|
||||
}
|
||||
|
||||
hashHostList(hosts: string[]): string {
|
||||
return crypto.createHash('sha256').update(hosts.slice().sort().join('\n')).digest('hex');
|
||||
}
|
||||
|
||||
signAttestation(payload: {
|
||||
nodeIdClaim: number;
|
||||
stack?: string;
|
||||
op: string;
|
||||
service?: string;
|
||||
sourceHash: string;
|
||||
referencedHostsHash: string;
|
||||
coveredHostsHash: string;
|
||||
actionSetHash: string;
|
||||
prepId?: string;
|
||||
}): string {
|
||||
const jti = crypto.randomBytes(16).toString('hex');
|
||||
return jwt.sign(
|
||||
{
|
||||
aud: ATTESTATION_AUD,
|
||||
nodeIdClaim: payload.nodeIdClaim,
|
||||
stack: payload.stack,
|
||||
op: payload.op,
|
||||
service: payload.service,
|
||||
sourceHash: payload.sourceHash,
|
||||
referencedHostsHash: payload.referencedHostsHash,
|
||||
coveredHostsHash: payload.coveredHostsHash,
|
||||
actionSetHash: payload.actionSetHash,
|
||||
prepId: payload.prepId,
|
||||
jti_t: jti,
|
||||
target_session_id: this.targetSessionId,
|
||||
},
|
||||
this.getJwtSecret(),
|
||||
{ expiresIn: ATTESTATION_TTL_SECONDS },
|
||||
);
|
||||
}
|
||||
|
||||
parseAttestation(token: string): jwt.JwtPayload {
|
||||
const decoded = jwt.verify(token, this.getJwtSecret(), { audience: ATTESTATION_AUD });
|
||||
if (typeof decoded === 'string') {
|
||||
throw new Error('Invalid attestation payload');
|
||||
}
|
||||
if (decoded.target_session_id !== this.targetSessionId) {
|
||||
throw new Error('Attestation session mismatch');
|
||||
}
|
||||
return decoded;
|
||||
}
|
||||
|
||||
private evictExpiredJtis(now = Date.now()): void {
|
||||
for (const [jti, expiresAt] of this.consumedJtis) {
|
||||
if (expiresAt <= now) {
|
||||
this.consumedJtis.delete(jti);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
consumeAttestationJti(jti: string, expiresAtMs?: number): void {
|
||||
const now = Date.now();
|
||||
this.evictExpiredJtis(now);
|
||||
if (this.consumedJtis.has(jti)) {
|
||||
throw new Error('Attestation already consumed');
|
||||
}
|
||||
if (this.consumedJtis.size >= this.maxConsumedJtis) {
|
||||
throw new Error('Attestation replay store at capacity');
|
||||
}
|
||||
const expiresAt = expiresAtMs ?? now + ATTESTATION_TTL_SECONDS * 1000;
|
||||
this.consumedJtis.set(jti, expiresAt);
|
||||
}
|
||||
|
||||
/** @deprecated Use parseAttestation at the middleware and consumeAttestationJti at the seam. */
|
||||
verifyAttestation(token: string): jwt.JwtPayload {
|
||||
const decoded = this.parseAttestation(token);
|
||||
const jti = decoded.jti_t;
|
||||
if (typeof jti !== 'string' || !jti) {
|
||||
throw new Error('Attestation missing jti');
|
||||
}
|
||||
this.consumeAttestationJti(jti);
|
||||
return decoded;
|
||||
}
|
||||
|
||||
async discoverOnTarget(request: RegistryDeliveryDiscoverRequest): Promise<RegistryDeliveryDiscoverResponse> {
|
||||
const nodeId = NodeRegistry.getInstance().getDefaultNodeId();
|
||||
if (
|
||||
request.sourceKind === 'restore-candidate'
|
||||
|| request.sourceKind === 'live-project'
|
||||
|| request.sourceKind === 'body-content'
|
||||
|| request.gitApply === true
|
||||
) {
|
||||
const stack = request.stack ?? request.stackName;
|
||||
if (typeof stack !== 'string' || !isValidStackName(stack)) {
|
||||
throw new Error('Invalid stack name');
|
||||
}
|
||||
request.stack = stack;
|
||||
}
|
||||
|
||||
let referencedHosts: string[] = [];
|
||||
let sourceHash = request.sourceHash;
|
||||
let prepId = request.prepId;
|
||||
|
||||
const prepared = await prepareSourceForDiscover(request);
|
||||
if (prepared) {
|
||||
prepId = prepared.prepId;
|
||||
sourceHash = prepared.sourceHash;
|
||||
const payloadPath = PreparedSourceStore.getInstance().peekPayloadPath(prepId);
|
||||
const discovery = discoverRegistryReferences(
|
||||
payloadPath,
|
||||
resolveComposeEnvForDiscovery(payloadPath, request.envVars),
|
||||
);
|
||||
referencedHosts = discovery.referencedHosts;
|
||||
} else if (request.sourceKind === 'body-content' && typeof request.composeContent === 'string') {
|
||||
const MAX_COMPOSE_CONTENT_BYTES = 2 * 1024 * 1024;
|
||||
if (Buffer.byteLength(request.composeContent, 'utf8') > MAX_COMPOSE_CONTENT_BYTES) {
|
||||
throw new Error('Compose content exceeds size limit');
|
||||
}
|
||||
const stack = request.stack ?? request.stackName;
|
||||
if (typeof stack !== 'string') {
|
||||
throw new Error('Invalid stack name');
|
||||
}
|
||||
const discovery = await resolveBlueprintPostApplyDiscovery(
|
||||
stack,
|
||||
request.composeContent,
|
||||
nodeId,
|
||||
);
|
||||
sourceHash = discovery.sourceHash;
|
||||
referencedHosts = discovery.referencedHosts;
|
||||
} else if (request.stack) {
|
||||
if (!isValidStackName(request.stack)) {
|
||||
throw new Error('Invalid stack name');
|
||||
}
|
||||
const { FileSystemService } = await import('./FileSystemService');
|
||||
const fs = FileSystemService.getInstance(nodeId);
|
||||
const baseResolved = path.resolve(fs.getBaseDir());
|
||||
const projectDir = path.resolve(baseResolved, request.stack);
|
||||
if (!projectDir.startsWith(baseResolved + path.sep)) {
|
||||
throw new Error('Invalid stack path');
|
||||
}
|
||||
if (!sourceHash || request.sourceKind === 'live-project') {
|
||||
sourceHash = hashProjectSource(projectDir);
|
||||
}
|
||||
const discovery = discoverRegistryReferences(
|
||||
projectDir,
|
||||
resolveComposeEnvForDiscovery(projectDir, request.envVars),
|
||||
);
|
||||
referencedHosts = discovery.referencedHosts;
|
||||
}
|
||||
|
||||
if (!sourceHash) {
|
||||
sourceHash = crypto.createHash('sha256').update('').digest('hex');
|
||||
}
|
||||
|
||||
const registry = RegistryService.getInstance();
|
||||
const coveredHosts: string[] = [];
|
||||
for (const host of referencedHosts) {
|
||||
const resolution = await registry.resolveDockerConfigForHostDetailed(host);
|
||||
if (resolution.state === 'unavailable') {
|
||||
throw new Error(`Registry credentials unavailable for ${host}`);
|
||||
}
|
||||
if (resolution.state === 'available') {
|
||||
coveredHosts.push(host);
|
||||
}
|
||||
}
|
||||
|
||||
const referencedHostsHash = this.hashHostList(referencedHosts);
|
||||
const coveredHostsHash = this.hashHostList(coveredHosts);
|
||||
const deliverySourceId = this.getDeliverySourceId();
|
||||
|
||||
const attestation = this.signAttestation({
|
||||
nodeIdClaim: nodeId,
|
||||
stack: request.stack,
|
||||
op: request.op,
|
||||
service: request.service,
|
||||
sourceHash,
|
||||
referencedHostsHash,
|
||||
coveredHostsHash,
|
||||
actionSetHash: request.actionSetHash,
|
||||
prepId,
|
||||
});
|
||||
|
||||
return {
|
||||
prepId,
|
||||
referencedHosts,
|
||||
coveredHosts,
|
||||
sourceHash,
|
||||
actionSetHash: request.actionSetHash,
|
||||
deliverySourceId,
|
||||
attestation,
|
||||
};
|
||||
}
|
||||
|
||||
async buildHubEnvelope(
|
||||
nodeId: number,
|
||||
discover: RegistryDeliveryDiscoverResponse,
|
||||
): Promise<RegistryDeliveryEnvelope | null> {
|
||||
const deltaHosts = discover.referencedHosts.filter(host => {
|
||||
return !discover.coveredHosts.includes(host);
|
||||
});
|
||||
|
||||
const registry = RegistryService.getInstance();
|
||||
const auths: RegistryDeliveryAuthEntry[] = [];
|
||||
|
||||
for (const host of deltaHosts) {
|
||||
const hubResolution: DockerConfigHostResolution = await registry.resolveDockerConfigForHostDetailed(host);
|
||||
if (hubResolution.state === 'unavailable') {
|
||||
throw new Error(`Hub registry credentials unavailable for ${host}`);
|
||||
}
|
||||
if (hubResolution.state === 'missing') {
|
||||
continue;
|
||||
}
|
||||
if (!hubResolution.auth) continue;
|
||||
auths.push({
|
||||
host,
|
||||
username: hubResolution.auth.username,
|
||||
password: hubResolution.auth.password,
|
||||
expiresAt: hubResolution.expiresAt,
|
||||
});
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const envelopeExp = now + ATTESTATION_TTL_SECONDS * 1000;
|
||||
const providerExpiries = auths.map(a => a.expiresAt).filter((v): v is number => typeof v === 'number');
|
||||
const notAfter = Math.min(envelopeExp, ...providerExpiries.length > 0 ? providerExpiries : [envelopeExp]);
|
||||
|
||||
return {
|
||||
attestation: discover.attestation,
|
||||
prepId: discover.prepId,
|
||||
auths,
|
||||
notAfter,
|
||||
deliverySourceId: discover.deliverySourceId,
|
||||
};
|
||||
}
|
||||
|
||||
isProxyTransportConfidential(nodeId: number): boolean {
|
||||
const node = NodeRegistry.getInstance().getNode(nodeId);
|
||||
if (!node?.api_url) return false;
|
||||
return node.api_url.trim().toLowerCase().startsWith('https://');
|
||||
}
|
||||
|
||||
isPilotTransportConfidential(socketEncrypted: boolean, forwardedProto: string | undefined, peerAddress: string | undefined): boolean {
|
||||
if (socketEncrypted) return true;
|
||||
if (forwardedProto?.toLowerCase() === 'https' && isTrustedProxyPeer(peerAddress)) {
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
async shouldAttemptDelivery(nodeId: number, confidential: boolean): Promise<boolean> {
|
||||
if (!confidential) return false;
|
||||
return remoteAdvertisesCapability(nodeId, REMOTE_REGISTRY_CREDENTIALS_CAPABILITY);
|
||||
}
|
||||
|
||||
isDeliveryEligibleRoute(method: string, apiPath: string): boolean {
|
||||
return classifyRegistryDeliveryOp(method, apiPath).eligible;
|
||||
}
|
||||
}
|
||||
@@ -111,7 +111,7 @@ export function hostFromStoredRegistry(reg: Pick<Registry, 'url' | 'type'>): str
|
||||
}
|
||||
|
||||
/** Normalize an image reference's host (the thing ImageUpdateService passes in). */
|
||||
function normalizeImageHost(host: string): string {
|
||||
export function normalizeImageHost(host: string): string {
|
||||
const lower = host.trim().toLowerCase();
|
||||
// Docker Hub aliases resolve to the same credential.
|
||||
if (lower === 'docker.io' || lower === 'registry-1.docker.io' || lower === '') {
|
||||
@@ -177,6 +177,14 @@ function httpGet(
|
||||
});
|
||||
}
|
||||
|
||||
export type DockerConfigHostState = 'missing' | 'available' | 'unavailable';
|
||||
|
||||
export interface DockerConfigHostResolution {
|
||||
state: DockerConfigHostState;
|
||||
auth?: { username: string; password: string };
|
||||
expiresAt?: number;
|
||||
}
|
||||
|
||||
// ─── Service ─────────────────────────────────────────────────────────────────
|
||||
|
||||
export class RegistryService {
|
||||
@@ -394,15 +402,17 @@ export class RegistryService {
|
||||
|
||||
/** Resolve a Docker config containing credentials for one registry host only. */
|
||||
public async resolveDockerConfigForHost(registryHost: string): Promise<ResolvedDockerConfig> {
|
||||
const auth = await this.getAuthForRegistry(registryHost);
|
||||
if (!auth) return { config: { auths: {} }, warnings: [] };
|
||||
const detailed = await this.resolveDockerConfigForHostDetailed(registryHost);
|
||||
if (detailed.state !== 'available' || !detailed.auth) {
|
||||
return { config: { auths: {} }, warnings: [] };
|
||||
}
|
||||
|
||||
const normalized = normalizeImageHost(registryHost);
|
||||
return {
|
||||
config: {
|
||||
auths: {
|
||||
[normalized]: {
|
||||
auth: Buffer.from(`${auth.username}:${auth.password}`).toString('base64'),
|
||||
auth: Buffer.from(`${detailed.auth.username}:${detailed.auth.password}`).toString('base64'),
|
||||
},
|
||||
},
|
||||
},
|
||||
@@ -410,6 +420,47 @@ export class RegistryService {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Tri-state host resolution for registry delivery. Target `unavailable`
|
||||
* means a configured row exists but credentials could not be resolved.
|
||||
*/
|
||||
public async resolveDockerConfigForHostDetailed(registryHost: string): Promise<DockerConfigHostResolution> {
|
||||
const db = DatabaseService.getInstance();
|
||||
const registries = db.getRegistries();
|
||||
const normalized = normalizeImageHost(registryHost);
|
||||
const match = registries.find(r => hostFromStoredRegistry(r) === normalized);
|
||||
|
||||
if (!match) {
|
||||
return { state: 'missing' };
|
||||
}
|
||||
|
||||
try {
|
||||
if (match.type === 'ecr') {
|
||||
const creds = await this.getEcrCredentials(match);
|
||||
const cached = this.ecrCache.get(match.id);
|
||||
const expiresAt = cached?.expiresAt ?? Date.now() + ECR_DEFAULT_TTL_MS;
|
||||
return {
|
||||
state: 'available',
|
||||
auth: { username: creds.username, password: creds.password },
|
||||
expiresAt,
|
||||
};
|
||||
}
|
||||
return {
|
||||
state: 'available',
|
||||
auth: {
|
||||
username: match.username,
|
||||
password: this.crypto.decrypt(match.secret),
|
||||
},
|
||||
};
|
||||
} catch (e) {
|
||||
console.warn(
|
||||
`[RegistryService] resolveDockerConfigForHostDetailed(${registryHost}) failed:`,
|
||||
sanitizeForLog((e as Error).message),
|
||||
);
|
||||
return { state: 'unavailable' };
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Resolve credentials for a specific registry row by ID only.
|
||||
|
||||
@@ -1158,6 +1158,35 @@ export class RollbackGenerationStore {
|
||||
}
|
||||
return Buffer.from(b64, 'base64');
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy present generation compose/project files into destDir for registry-delivery
|
||||
* preparation. Preserves stack-relative paths under destDir.
|
||||
*/
|
||||
static async copyPresentFilesToDir(
|
||||
nodeId: number,
|
||||
stackName: string,
|
||||
generationId: string,
|
||||
destDir: string,
|
||||
): Promise<void> {
|
||||
assertSafeStackName(stackName);
|
||||
assertSafeGenerationId(generationId);
|
||||
const genDir = this.getGenerationDir(nodeId, stackName, generationId);
|
||||
const manifest = await this.readAndVerifyGeneration(genDir);
|
||||
await mkdirPrivate(destDir);
|
||||
for (const entry of manifest.entries) {
|
||||
if (entry.state !== 'present') continue;
|
||||
const rel = posixRel(entry.relativePath);
|
||||
const dest = path.resolve(destDir, rel);
|
||||
const destRoot = path.resolve(destDir);
|
||||
if (!dest.startsWith(destRoot + path.sep) && dest !== destRoot) {
|
||||
throw Object.assign(new Error('Path escapes restore staging directory'), { code: 'INVALID_PATH' });
|
||||
}
|
||||
await mkdirPrivate(path.dirname(dest));
|
||||
const content = await this.readPresentEntryBytes(genDir, entry);
|
||||
await writePrivate(dest, content);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export { getBackupBaseDir, getDataDir };
|
||||
|
||||
@@ -32,6 +32,7 @@ import type { ScanAllNodeImagesResult } from './TrivyService';
|
||||
import TrivyInstaller from './TrivyInstaller';
|
||||
import { CloudBackupService } from './CloudBackupService';
|
||||
import { buildSystemPolicyGateOptions } from '../helpers/policyGate';
|
||||
import { prepareOutboundRegistryDeliveryBody } from '../helpers/registryDeliveryOutbound';
|
||||
import { filterContainersByComposeService } from '../helpers/composeServiceMatch';
|
||||
import { excludeSelfContainers } from '../helpers/excludeSelfContainers';
|
||||
import { enforcePolicyPreDeploy } from './PolicyEnforcement';
|
||||
@@ -1245,10 +1246,20 @@ export class SchedulerService {
|
||||
const proxyTarget = this.requireRemoteProxyTarget(nodeId);
|
||||
const baseUrl = proxyTarget.apiUrl.replace(/\/$/, '');
|
||||
const proxyHeaders = LicenseService.getInstance().getProxyHeaders();
|
||||
const apiPath = `/api/stacks/${routeSuffix}`;
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[SchedulerService:debug] postToRemoteStack: node=${nodeId} route=${routeSuffix}`);
|
||||
}
|
||||
try {
|
||||
const augmented = await prepareOutboundRegistryDeliveryBody({
|
||||
method: 'POST',
|
||||
apiPath,
|
||||
nodeId,
|
||||
body: {},
|
||||
});
|
||||
if (!augmented.ok) {
|
||||
throw new Error(augmented.error);
|
||||
}
|
||||
const response = await fetch(`${baseUrl}/api/stacks/${routeSuffix}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
@@ -1257,6 +1268,7 @@ export class SchedulerService {
|
||||
[PROXY_TIER_HEADER]: proxyHeaders.tier,
|
||||
...extraHeaders,
|
||||
},
|
||||
body: JSON.stringify(augmented.body),
|
||||
signal: AbortSignal.timeout(300_000),
|
||||
});
|
||||
if (!response.ok) {
|
||||
|
||||
@@ -21,10 +21,16 @@ export function stackOpSkipMessage(stackName: string, existingAction: StackOpAct
|
||||
return `Skipped "${stackName}": another operation (${existingAction}) is already in progress.`;
|
||||
}
|
||||
|
||||
export interface StackOpLockContext {
|
||||
opId?: string;
|
||||
kind?: string;
|
||||
}
|
||||
|
||||
export interface StackOpLock {
|
||||
action: StackOpAction;
|
||||
startedAt: number;
|
||||
user: string;
|
||||
context?: StackOpLockContext;
|
||||
}
|
||||
|
||||
interface AcquireSuccess {
|
||||
@@ -60,11 +66,12 @@ export class StackOpLockService {
|
||||
stackName: string,
|
||||
action: StackOpAction,
|
||||
user: string,
|
||||
context?: StackOpLockContext,
|
||||
): AcquireResult {
|
||||
const k = this.key(nodeId, stackName);
|
||||
const existing = this.locks.get(k);
|
||||
if (existing) return { acquired: false, existing };
|
||||
this.locks.set(k, { action, startedAt: Date.now(), user });
|
||||
this.locks.set(k, { action, startedAt: Date.now(), user, context });
|
||||
return { acquired: true };
|
||||
}
|
||||
|
||||
@@ -114,8 +121,9 @@ export class StackOpLockService {
|
||||
action: StackOpAction,
|
||||
user: string,
|
||||
fn: () => Promise<T>,
|
||||
context?: StackOpLockContext,
|
||||
): Promise<{ ran: true; result: T } | { ran: false; existing: StackOpLock }> {
|
||||
const acquired = this.tryAcquire(nodeId, stackName, action, user);
|
||||
const acquired = this.tryAcquire(nodeId, stackName, action, user, context);
|
||||
if (!acquired.acquired) return { ran: false, existing: acquired.existing };
|
||||
try {
|
||||
const result = await fn();
|
||||
|
||||
@@ -3,6 +3,7 @@ import YAML from 'yaml';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
import { CacheService } from './CacheService';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
import { isValidStackName } from '../utils/validation';
|
||||
|
||||
|
||||
interface TemplateEnv {
|
||||
@@ -365,6 +366,9 @@ export class TemplateService {
|
||||
}
|
||||
|
||||
public generateComposeFromTemplate(template: Template, serviceName: string): string {
|
||||
if (!isValidStackName(serviceName)) {
|
||||
throw new Error('Invalid service name');
|
||||
}
|
||||
const service: ComposeServiceDefinition = { restart: 'unless-stopped' };
|
||||
|
||||
if (template.image) {
|
||||
|
||||
@@ -13,6 +13,7 @@ import { getErrorMessage } from '../utils/errors';
|
||||
import { redactSensitiveText } from '../utils/safeLog';
|
||||
import { isValidStackName } from '../utils/validation';
|
||||
import { assertPolicyGateAllows, buildSystemPolicyGateOptions } from '../helpers/policyGate';
|
||||
import { prepareOutboundRegistryDeliveryBody } from '../helpers/registryDeliveryOutbound';
|
||||
|
||||
type ExecutionResult = { success: boolean; error?: string; duration_ms: number };
|
||||
type ExecutionStatus = 'success' | 'failure';
|
||||
@@ -335,10 +336,42 @@ export class WebhookService {
|
||||
|
||||
// Build URL from validated, server-controlled components.
|
||||
const url = `${protocol}//${host}/api/stacks/${encodeURIComponent(stackName)}/${endpoint}`;
|
||||
const apiPath = `/api/stacks/${encodeURIComponent(stackName)}/${endpoint}`;
|
||||
let bodyToSend = body;
|
||||
if (method === 'POST' && body !== undefined) {
|
||||
const bodyRecord = typeof body === 'object' && body !== null && !Array.isArray(body)
|
||||
? body as Record<string, unknown>
|
||||
: {};
|
||||
const augmented = await prepareOutboundRegistryDeliveryBody({
|
||||
method,
|
||||
apiPath,
|
||||
nodeId,
|
||||
body: bodyRecord,
|
||||
});
|
||||
if (!augmented.ok) {
|
||||
const err = new Error(augmented.error);
|
||||
(err as { status?: number }).status = augmented.status;
|
||||
throw err;
|
||||
}
|
||||
bodyToSend = augmented.body;
|
||||
} else if (method === 'POST') {
|
||||
const augmented = await prepareOutboundRegistryDeliveryBody({
|
||||
method,
|
||||
apiPath,
|
||||
nodeId,
|
||||
body: {},
|
||||
});
|
||||
if (!augmented.ok) {
|
||||
const err = new Error(augmented.error);
|
||||
(err as { status?: number }).status = augmented.status;
|
||||
throw err;
|
||||
}
|
||||
bodyToSend = augmented.body;
|
||||
}
|
||||
return await fetch(url, {
|
||||
method,
|
||||
headers,
|
||||
body: method === 'GET' || body === undefined ? undefined : JSON.stringify(body),
|
||||
body: method === 'GET' || bodyToSend === undefined ? undefined : JSON.stringify(bodyToSend),
|
||||
signal: controller.signal,
|
||||
});
|
||||
} catch (err) {
|
||||
|
||||
@@ -0,0 +1,219 @@
|
||||
import crypto from 'crypto';
|
||||
import fs from 'fs';
|
||||
import os from 'os';
|
||||
import path from 'path';
|
||||
import { ensureTrustedRoot, validateTrustedRoot } from '../helpers/privateRootValidator';
|
||||
|
||||
export const PREPARED_SOURCE_MARKER_FILE = '.sencho-prepared-source';
|
||||
export const PREPARED_SOURCE_PARENT_PREFIX = 'sencho-registry-prepared-';
|
||||
const DEFAULT_TTL_MS = 900_000;
|
||||
|
||||
export type PreparedSourceState = 'prepared' | 'claimed' | 'finalized';
|
||||
|
||||
export interface PreparedSourceEntry {
|
||||
prepId: string;
|
||||
sourceKind: string;
|
||||
sourceHash: string;
|
||||
dirPath: string;
|
||||
state: PreparedSourceState;
|
||||
createdAt: number;
|
||||
expiresAt: number;
|
||||
}
|
||||
|
||||
function deliverySourceHash(deliverySourceId: string): string {
|
||||
return crypto.createHash('sha256').update(`prepared-source:${deliverySourceId}`).digest('hex');
|
||||
}
|
||||
|
||||
export function getPreparedSourceRootPath(deliverySourceId: string): string {
|
||||
return path.join(os.tmpdir(), `${PREPARED_SOURCE_PARENT_PREFIX}${deliverySourceHash(deliverySourceId)}`);
|
||||
}
|
||||
|
||||
function publishMarker(childDir: string, sourceKind: string): void {
|
||||
const markerPath = path.join(childDir, PREPARED_SOURCE_MARKER_FILE);
|
||||
const fd = fs.openSync(markerPath, 'wx', 0o600);
|
||||
try {
|
||||
fs.writeSync(fd, `${sourceKind}\n`);
|
||||
fs.fsyncSync(fd);
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
}
|
||||
}
|
||||
|
||||
export class PreparedSourceStore {
|
||||
private static instance: PreparedSourceStore | null = null;
|
||||
private entries = new Map<string, PreparedSourceEntry>();
|
||||
private expiryTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private deliverySourceId: string | null = null;
|
||||
|
||||
static getInstance(): PreparedSourceStore {
|
||||
if (!this.instance) this.instance = new PreparedSourceStore();
|
||||
return this.instance;
|
||||
}
|
||||
|
||||
configure(deliverySourceId: string): void {
|
||||
this.deliverySourceId = deliverySourceId;
|
||||
}
|
||||
|
||||
start(): void {
|
||||
if (this.expiryTimer) return;
|
||||
this.expiryTimer = setInterval(() => this.expireStaleEntries(), 60_000);
|
||||
if (typeof this.expiryTimer.unref === 'function') {
|
||||
this.expiryTimer.unref();
|
||||
}
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
if (this.expiryTimer) {
|
||||
clearInterval(this.expiryTimer);
|
||||
this.expiryTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private requireDeliverySourceId(): string {
|
||||
if (!this.deliverySourceId) {
|
||||
throw new Error('PreparedSourceStore is not configured');
|
||||
}
|
||||
return this.deliverySourceId;
|
||||
}
|
||||
|
||||
private childPath(prepId: string): string {
|
||||
const root = getPreparedSourceRootPath(this.requireDeliverySourceId());
|
||||
return path.join(root, prepId);
|
||||
}
|
||||
|
||||
async prepareFromDirectory(
|
||||
sourceKind: string,
|
||||
sourceHash: string,
|
||||
stagingDir: string,
|
||||
): Promise<PreparedSourceEntry> {
|
||||
const deliverySourceId = this.requireDeliverySourceId();
|
||||
const rootPath = getPreparedSourceRootPath(deliverySourceId);
|
||||
const rootValidation = ensureTrustedRoot({ rootPath, kind: 'prepared-source' });
|
||||
if (!rootValidation.ok) {
|
||||
throw new Error(rootValidation.reason);
|
||||
}
|
||||
|
||||
const prepId = crypto.randomBytes(16).toString('hex');
|
||||
const childDir = path.join(rootPath, prepId);
|
||||
fs.mkdirSync(childDir, { mode: 0o700 });
|
||||
|
||||
try {
|
||||
publishMarker(childDir, sourceKind);
|
||||
const payloadDir = path.join(childDir, 'payload');
|
||||
await fs.promises.rename(stagingDir, payloadDir);
|
||||
} catch (error) {
|
||||
try { await fs.promises.rm(childDir, { recursive: true, force: true }); } catch { /* ignore */ }
|
||||
throw error;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const entry: PreparedSourceEntry = {
|
||||
prepId,
|
||||
sourceKind,
|
||||
sourceHash,
|
||||
dirPath: childDir,
|
||||
state: 'prepared',
|
||||
createdAt: now,
|
||||
expiresAt: now + DEFAULT_TTL_MS,
|
||||
};
|
||||
this.entries.set(prepId, entry);
|
||||
return entry;
|
||||
}
|
||||
|
||||
getEntry(prepId: string): PreparedSourceEntry | undefined {
|
||||
return this.entries.get(prepId);
|
||||
}
|
||||
|
||||
claim(prepId: string): PreparedSourceEntry {
|
||||
const entry = this.entries.get(prepId);
|
||||
if (!entry || entry.state !== 'prepared') {
|
||||
throw new Error('Prepared source is not available for claim');
|
||||
}
|
||||
if (entry.expiresAt <= Date.now()) {
|
||||
throw new Error('Prepared source expired');
|
||||
}
|
||||
entry.state = 'claimed';
|
||||
return entry;
|
||||
}
|
||||
|
||||
peekPayloadPath(prepId: string): string {
|
||||
const entry = this.entries.get(prepId);
|
||||
if (!entry || entry.state === 'finalized') {
|
||||
throw new Error('Prepared source not found');
|
||||
}
|
||||
if (entry.expiresAt <= Date.now()) {
|
||||
throw new Error('Prepared source expired');
|
||||
}
|
||||
return path.join(entry.dirPath, 'payload');
|
||||
}
|
||||
|
||||
finalize(prepId: string): void {
|
||||
const entry = this.entries.get(prepId);
|
||||
if (!entry) return;
|
||||
entry.state = 'finalized';
|
||||
try {
|
||||
fs.rmSync(entry.dirPath, { recursive: true, force: true });
|
||||
} catch {
|
||||
/* best effort */
|
||||
}
|
||||
this.entries.delete(prepId);
|
||||
}
|
||||
|
||||
getPayloadPath(prepId: string): string {
|
||||
const entry = this.entries.get(prepId);
|
||||
if (!entry) {
|
||||
throw new Error('Prepared source not found');
|
||||
}
|
||||
return path.join(entry.dirPath, 'payload');
|
||||
}
|
||||
|
||||
private expireStaleEntries(): void {
|
||||
const now = Date.now();
|
||||
for (const [prepId, entry] of this.entries) {
|
||||
if (entry.expiresAt <= now) {
|
||||
try {
|
||||
fs.rmSync(entry.dirPath, { recursive: true, force: true });
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
this.entries.delete(prepId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async sweepOrphans(deliverySourceId: string): Promise<string[]> {
|
||||
const rootPath = getPreparedSourceRootPath(deliverySourceId);
|
||||
const swept: string[] = [];
|
||||
const validation = validateTrustedRoot({ rootPath, kind: 'prepared-source' });
|
||||
if (!validation.ok) {
|
||||
return swept;
|
||||
}
|
||||
|
||||
let entries: string[];
|
||||
try {
|
||||
entries = await fs.promises.readdir(rootPath);
|
||||
} catch {
|
||||
return swept;
|
||||
}
|
||||
|
||||
for (const entry of entries) {
|
||||
const childPath = path.join(rootPath, entry);
|
||||
let stat: fs.Stats;
|
||||
try {
|
||||
stat = await fs.promises.lstat(childPath);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (!stat.isDirectory() || stat.isSymbolicLink()) continue;
|
||||
const markerPath = path.join(childPath, PREPARED_SOURCE_MARKER_FILE);
|
||||
if (!fs.existsSync(markerPath)) continue;
|
||||
try {
|
||||
await fs.promises.rm(childPath, { recursive: true, force: true });
|
||||
swept.push(entry);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
return swept;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { parseImageRef } from './registry-api';
|
||||
import { extractImagesFromCompose } from './ImageUpdateService';
|
||||
import { normalizeImageHost } from './RegistryService';
|
||||
|
||||
const MAX_DOCKERFILE_BYTES = 1_048_576;
|
||||
|
||||
export interface RegistryReferenceDiscoveryResult {
|
||||
referencedHosts: string[];
|
||||
}
|
||||
|
||||
function hostFromImageRef(imageRef: string): string | null {
|
||||
const parsed = parseImageRef(imageRef);
|
||||
if (!parsed) return null;
|
||||
return normalizeImageHost(parsed.registry);
|
||||
}
|
||||
|
||||
function parseDockerfileReferences(content: string): string[] {
|
||||
const hosts = new Set<string>();
|
||||
const lines = content.split('\n');
|
||||
for (const rawLine of lines) {
|
||||
const line = rawLine.trim();
|
||||
if (!line || line.startsWith('#')) continue;
|
||||
|
||||
const fromMatch = /^FROM\s+(--platform=[^\s]+\s+)?([^\s]+)/i.exec(line);
|
||||
if (fromMatch?.[2]) {
|
||||
const host = hostFromImageRef(fromMatch[2]);
|
||||
if (host) hosts.add(host);
|
||||
}
|
||||
|
||||
const copyFromMatch = /^COPY\s+--from=([^\s]+)/i.exec(line);
|
||||
if (copyFromMatch?.[1] && !/^\d+$/.test(copyFromMatch[1])) {
|
||||
const host = hostFromImageRef(copyFromMatch[1]);
|
||||
if (host) hosts.add(host);
|
||||
}
|
||||
}
|
||||
return [...hosts];
|
||||
}
|
||||
|
||||
function readRegularFileSync(filePath: string, baseResolved: string): Buffer | null {
|
||||
const resolved = path.resolve(filePath);
|
||||
if (!resolved.startsWith(baseResolved + path.sep)) return null;
|
||||
const fd = fs.openSync(resolved, 'r');
|
||||
try {
|
||||
const stat = fs.fstatSync(fd);
|
||||
if (!stat.isFile()) return null;
|
||||
const buf = Buffer.alloc(stat.size);
|
||||
fs.readSync(fd, buf, 0, stat.size, 0);
|
||||
return buf;
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
}
|
||||
}
|
||||
|
||||
function discoverFromComposeFile(baseResolved: string, fileName: string, envVars: Record<string, string>): string[] {
|
||||
const safePath = path.resolve(baseResolved, fileName);
|
||||
if (!safePath.startsWith(baseResolved + path.sep)) {
|
||||
return [];
|
||||
}
|
||||
const content = readRegularFileSync(safePath, baseResolved);
|
||||
if (!content) return [];
|
||||
const images = extractImagesFromCompose(content.toString('utf8'), envVars);
|
||||
const hosts = new Set<string>();
|
||||
for (const image of images) {
|
||||
const host = hostFromImageRef(image);
|
||||
if (host) hosts.add(host);
|
||||
}
|
||||
return [...hosts];
|
||||
}
|
||||
|
||||
function discoverDockerfiles(baseResolved: string): string[] {
|
||||
const hosts = new Set<string>();
|
||||
const stack: string[] = [baseResolved];
|
||||
while (stack.length > 0) {
|
||||
const current = stack.pop();
|
||||
if (!current) continue;
|
||||
let entries: fs.Dirent[];
|
||||
try {
|
||||
entries = fs.readdirSync(current, { withFileTypes: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
for (const entry of entries) {
|
||||
const full = path.resolve(current, entry.name);
|
||||
if (!full.startsWith(baseResolved + path.sep)) continue;
|
||||
if (entry.isDirectory()) {
|
||||
stack.push(full);
|
||||
continue;
|
||||
}
|
||||
if (!entry.isFile()) continue;
|
||||
const lower = entry.name.toLowerCase();
|
||||
if (lower !== 'dockerfile' && !lower.startsWith('dockerfile.')) continue;
|
||||
const fd = fs.openSync(full, 'r');
|
||||
try {
|
||||
const stat = fs.fstatSync(fd);
|
||||
if (stat.size > MAX_DOCKERFILE_BYTES) {
|
||||
throw new Error(`Dockerfile exceeds size limit: ${entry.name}`);
|
||||
}
|
||||
const buf = Buffer.alloc(stat.size);
|
||||
fs.readSync(fd, buf, 0, stat.size, 0);
|
||||
for (const host of parseDockerfileReferences(buf.toString('utf8'))) {
|
||||
hosts.add(host);
|
||||
}
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
}
|
||||
}
|
||||
}
|
||||
return [...hosts];
|
||||
}
|
||||
|
||||
export function discoverRegistryReferencesFromComposeContent(
|
||||
composeContent: string,
|
||||
envVars: Record<string, string> = {},
|
||||
): RegistryReferenceDiscoveryResult {
|
||||
const hosts = new Set<string>();
|
||||
for (const image of extractImagesFromCompose(composeContent, envVars)) {
|
||||
const host = hostFromImageRef(image);
|
||||
if (host) hosts.add(host);
|
||||
}
|
||||
return { referencedHosts: [...hosts].sort() };
|
||||
}
|
||||
|
||||
/**
|
||||
* Discover registry hosts referenced by compose files and Dockerfiles in a
|
||||
* project directory. Returns referenced hosts, not proven-private hosts.
|
||||
*/
|
||||
export function discoverRegistryReferences(
|
||||
projectDir: string,
|
||||
envVars: Record<string, string> = {},
|
||||
): RegistryReferenceDiscoveryResult {
|
||||
const hosts = new Set<string>();
|
||||
const baseResolved = path.resolve(projectDir);
|
||||
|
||||
const composeNames = ['compose.yaml', 'compose.yml', 'docker-compose.yaml', 'docker-compose.yml'];
|
||||
for (const name of composeNames) {
|
||||
const composePath = path.resolve(baseResolved, name);
|
||||
if (!composePath.startsWith(baseResolved + path.sep)) continue;
|
||||
if (!fs.existsSync(composePath)) continue;
|
||||
for (const host of discoverFromComposeFile(baseResolved, name, envVars)) {
|
||||
hosts.add(host);
|
||||
}
|
||||
}
|
||||
|
||||
for (const host of discoverDockerfiles(baseResolved)) {
|
||||
hosts.add(host);
|
||||
}
|
||||
|
||||
return { referencedHosts: [...hosts].sort() };
|
||||
}
|
||||
|
||||
export { parseDockerfileReferences };
|
||||
Reference in New Issue
Block a user