Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
178 changes: 71 additions & 107 deletions design-qa.md

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions openclaw-plugin/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ export class PluginConfig {
};
semantic: {
enabled: boolean;
cacheEnabled: boolean;
mode: SemanticJudgeMode;
judgeToolCalls: boolean;
judgeMessages: boolean;
Expand Down Expand Up @@ -224,6 +225,7 @@ export class PluginConfig {
};
this.semantic = {
enabled: false,
cacheEnabled: true,
mode: "risk-tiered",
judgeToolCalls: true,
judgeMessages: false,
Expand Down Expand Up @@ -355,6 +357,7 @@ export class PluginConfig {
const semantic = objectAt(obj, "semantic");
if (semantic) {
config.semantic.enabled = readBoolean(semantic.enabled, config.semantic.enabled);
config.semantic.cacheEnabled = readBoolean(semantic.cacheEnabled, config.semantic.cacheEnabled);
const mode = readString(semantic.mode, config.semantic.mode);
if (mode === "off" || mode === "risk-tiered" || mode === "full") {
config.semantic.mode = mode;
Expand Down
30 changes: 27 additions & 3 deletions openclaw-plugin/core/persistence/event-writer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ export type EventWriterOptions = {
export class EventWriter<T extends WritableEvent> {
private readonly recordsPath: string;
private readonly batchIntervalMs: number;
private readonly batchSize: number;
private batchSize: number;
private readonly maxQueue: number;
private readonly compactEvery: number;
private readonly maxRecords: number;
Expand Down Expand Up @@ -43,6 +43,12 @@ export class EventWriter<T extends WritableEvent> {
return true;
}

setBatchSize(batchSize: number): void {
if (!Number.isFinite(batchSize)) return;
this.batchSize = Math.max(1, Math.min(1000, Math.trunc(batchSize)));
if (this.queue.length >= this.batchSize) this.scheduleFlush(0);
}

flush(): Promise<void> {
this.clearTimer();
let completion = this.operation;
Expand Down Expand Up @@ -79,6 +85,13 @@ export class EventWriter<T extends WritableEvent> {
return task;
}

pruneBefore(cutoff: string): Promise<void> {
const flushed = this.flush();
const task = flushed.then(() => compactFile(this.recordsPath, this.maxRecords, cutoff));
this.operation = task.catch(() => undefined);
return task;
}

async close(): Promise<void> {
this.closing = true;
await this.flush();
Expand Down Expand Up @@ -122,11 +135,22 @@ async function needsLeadingNewline(recordsPath: string): Promise<boolean> {
}
}

async function compactFile(recordsPath: string, maxRecords: number): Promise<void> {
async function compactFile(recordsPath: string, maxRecords: number, cutoff?: string): Promise<void> {
const tmpPath = `${recordsPath}.tmp`;
try {
const content = await readFile(recordsPath, "utf8");
const lines = content.split(/\r?\n/).filter(Boolean).slice(-maxRecords);
const cutoffMs = cutoff ? new Date(cutoff).getTime() : Number.NaN;
const lines = content.split(/\r?\n/).filter(Boolean).filter((line) => {
if (!Number.isFinite(cutoffMs)) return true;
try {
const parsed = JSON.parse(line) as { created_at?: unknown };
const createdAt = new Date(String(parsed.created_at || "")).getTime();
return !Number.isFinite(createdAt) || createdAt >= cutoffMs;
} catch {
// Preserve malformed lines for forensic review; normal compaction still bounds the file.
return true;
}
}).slice(-maxRecords);
await writeFile(tmpPath, `${lines.join("\n")}${lines.length ? "\n" : ""}`, { encoding: "utf8", mode: 0o600 });
await rename(tmpPath, recordsPath);
if (process.platform !== "win32") await chmod(recordsPath, 0o600);
Expand Down
97 changes: 96 additions & 1 deletion openclaw-plugin/core/records.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,13 @@ export type AgentSentryRecord = {
summary: string;
payload: Record<string, unknown>;
created_at: string;
previous_hash?: string;
event_hash?: string;
};

export const AUDIT_GENESIS_HASH = createHash("sha256").update("agentsentry-audit-genesis", "utf8").digest("hex");
export type RecordListener = (record: AgentSentryRecord) => void | Promise<void>;

export class RecordStore {
readonly stateDir: string;
readonly dataDir: string;
Expand All @@ -33,6 +38,9 @@ export class RecordStore {
private resetPending = false;
private countCache: { size: number; mtimeMs: number; count: number } | null = null;
private statsCache = new Map<number, { size: number; mtimeMs: number; value: Record<string, unknown> }>();
private auditHashChainEnabled = true;
private auditHead: string;
private listeners = new Set<RecordListener>();

constructor(config: PluginConfig) {
this.stateDir = config.storage.stateDir || process.env.OPENCLAW_STATE_DIR?.trim() || join(homedir(), ".openclaw");
Expand All @@ -45,6 +53,21 @@ export class RecordStore {
restrictAuditPermissions(this.dataDir, this.recordsPath);
this.knownCount = countValidRecords(this.recordsPath);
this.writer = new EventWriter(this.recordsPath, { maxRecords: this.maxRecords });
this.auditHead = latestAuditHash(this.recordsPath) || AUDIT_GENESIS_HASH;
}

configureAudit(options: { hashChain?: boolean; batchSize?: number } = {}): void {
if (typeof options.hashChain === "boolean") this.auditHashChainEnabled = options.hashChain;
if (options.batchSize !== undefined) this.writer.setBatchSize(options.batchSize);
}

auditConfig(): { hashChain: boolean } {
return { hashChain: this.auditHashChainEnabled };
}

subscribe(listener: RecordListener): () => void {
this.listeners.add(listener);
return () => this.listeners.delete(listener);
}

add(input: Omit<AgentSentryRecord, "id" | "created_at"> & { id?: string; created_at?: string }): AgentSentryRecord {
Expand All @@ -60,13 +83,23 @@ export class RecordStore {
summary: input.summary,
payload: input.payload,
};
const persistedRecord = recordForStorage(record, this.previewChars);
const persistedRecord = withAuditHash(recordForStorage(record, this.previewChars), this.auditHashChainEnabled, this.auditHead);
if (this.writer.enqueue(persistedRecord)) {
if (persistedRecord.event_hash) this.auditHead = persistedRecord.event_hash;
this.recentRecords.push(persistedRecord);
if (this.recentRecords.length > this.maxRecords) this.recentRecords = this.recentRecords.slice(-this.maxRecords);
this.knownCount = Math.min(this.maxRecords, this.knownCount + 1);
this.countCache = null;
this.statsCache.clear();
for (const listener of this.listeners) {
queueMicrotask(() => {
try {
void Promise.resolve(listener(persistedRecord)).catch(() => undefined);
} catch {
// Audit persistence must not depend on optional observers.
}
});
}
}
return persistedRecord;
}
Expand Down Expand Up @@ -167,6 +200,7 @@ export class RecordStore {
this.knownCount = 0;
this.countCache = null;
this.statsCache.clear();
this.auditHead = AUDIT_GENESIS_HASH;
this.resetPending = true;
void this.writer.reset().finally(() => {
this.resetPending = false;
Expand All @@ -177,6 +211,18 @@ export class RecordStore {
return this.writer.compact();
}

pruneBefore(cutoff: string): Promise<void> {
return this.writer.pruneBefore(cutoff).then(() => {
const cutoffMs = new Date(cutoff).getTime();
if (Number.isFinite(cutoffMs)) {
this.recentRecords = this.recentRecords.filter((record) => new Date(record.created_at).getTime() >= cutoffMs);
}
this.knownCount = countValidRecords(this.recordsPath);
this.countCache = null;
this.statsCache.clear();
});
}

async flush(): Promise<void> {
await this.writer.flush();
this.recentRecords = [];
Expand Down Expand Up @@ -226,6 +272,55 @@ function recordForStorage(record: AgentSentryRecord, previewChars: number): Agen
};
}

function withAuditHash(record: AgentSentryRecord, enabled: boolean, previousHash: string): AgentSentryRecord {
const next: AgentSentryRecord = { ...record };
delete next.previous_hash;
delete next.event_hash;
if (next.payload && typeof next.payload === "object") {
const payload = { ...next.payload };
delete payload.previous_hash;
delete payload.event_hash;
next.payload = payload;
}
if (!enabled) return next;
const previous = previousHash || AUDIT_GENESIS_HASH;
const eventHash = computeAuditEventHash(next, previous);
return { ...next, previous_hash: previous, event_hash: eventHash };
}

export function computeAuditEventHash(record: Pick<AgentSentryRecord, "id" | "run_id" | "session_key" | "type" | "layer" | "severity" | "title" | "summary" | "payload" | "created_at">, previousHash: string): string {
const payload = { ...(record.payload || {}) };
delete payload.previous_hash;
delete payload.event_hash;
const canonical = {
id: record.id,
run_id: record.run_id,
session_key: record.session_key,
type: record.type,
layer: record.layer,
severity: record.severity,
title: record.title,
summary: record.summary,
payload,
created_at: record.created_at,
};
return createHash("sha256").update(previousHash, "utf8").update("\n", "utf8").update(stableJson(canonical), "utf8").digest("hex");
}

function stableJson(value: unknown): string {
if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;
if (value && typeof value === "object") {
const record = value as Record<string, unknown>;
return `{${Object.keys(record).sort().map((key) => `${JSON.stringify(key)}:${stableJson(record[key])}`).join(",")}}`;
}
return JSON.stringify(value) ?? "null";
}

function latestAuditHash(path: string): string | null {
const latest = readTailRecords(path, 1)[0];
return latest?.event_hash || (typeof latest?.payload?.event_hash === "string" ? latest.payload.event_hash : null);
}

function readTailRecords(path: string, limit: number): AgentSentryRecord[] {
if (!existsSync(path)) return [];
const stat = statSync(path);
Expand Down
20 changes: 12 additions & 8 deletions openclaw-plugin/core/semantic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,21 +68,25 @@ export async function semanticJudgeAmbiguousAction(input: {
if (!config.semantic.enabled || !config.semantic.judgeToolCalls || config.semantic.mode === "off") return [];
if (input.preliminary.deterministic_disposition !== "ambiguous") return [];
const cacheKey = semanticActionCacheKey(input.taskSpec, input.action);
const cached = semanticActionCache.get(cacheKey);
if (cached?.expiresAt && cached.expiresAt > Date.now()) return markSemanticCacheHit(cached.findings, cacheKey);
if (cached) semanticActionCache.delete(cacheKey);
if (config.semantic.cacheEnabled) {
const cached = semanticActionCache.get(cacheKey);
if (cached?.expiresAt && cached.expiresAt > Date.now()) return markSemanticCacheHit(cached.findings, cacheKey);
if (cached) semanticActionCache.delete(cacheKey);
}
const inflight = semanticActionInflight.get(cacheKey);
if (inflight) return markSemanticCacheHit(await inflight, cacheKey);

const request = judgeAmbiguousAction(input, config, cacheKey);
semanticActionInflight.set(cacheKey, request);
try {
const findings = await request;
semanticActionCache.set(cacheKey, { expiresAt: Date.now() + SEMANTIC_ACTION_CACHE_TTL_MS, findings: structuredClone(findings) });
while (semanticActionCache.size > SEMANTIC_ACTION_CACHE_MAX) {
const oldest = semanticActionCache.keys().next().value;
if (oldest === undefined) break;
semanticActionCache.delete(oldest);
if (config.semantic.cacheEnabled) {
semanticActionCache.set(cacheKey, { expiresAt: Date.now() + SEMANTIC_ACTION_CACHE_TTL_MS, findings: structuredClone(findings) });
while (semanticActionCache.size > SEMANTIC_ACTION_CACHE_MAX) {
const oldest = semanticActionCache.keys().next().value;
if (oldest === undefined) break;
semanticActionCache.delete(oldest);
}
}
return findings;
} finally {
Expand Down
21 changes: 20 additions & 1 deletion openclaw-plugin/core/ssrf-http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,18 @@ export async function safeHttpGet(input: URL | string, options: SafeHttpOptions
throw new Error("HTTP redirect limit exceeded");
}

export async function safeHttpPost(input: URL | string, body: unknown, options: SafeHttpOptions = {}): Promise<SafeHttpResponse> {
const maxBytes = positiveInt(options.maxBytes, DEFAULT_MAX_BYTES);
const timeoutMs = positiveInt(options.timeoutMs, DEFAULT_TIMEOUT_MS);
const url = typeof input === "string" ? new URL(input) : new URL(input.toString());
const allowedHosts = options.allowedHosts || [];
validateHttpUrl(url, allowedHosts);
const resolved = await resolvePublicAddress(url.hostname, isHostAllowlisted(url, allowedHosts));
const requestBody = typeof body === "string" ? body : JSON.stringify(body);
const response = await requestPinned(url, resolved, maxBytes, timeoutMs, "POST", requestBody);
return { ...response, finalUrl: url.toString(), redirects: 0 };
}

export function isForbiddenIpAddress(address: string): boolean {
const family = isIP(stripIpv6Brackets(address));
if (family === 4) return forbiddenIpv4(stripIpv6Brackets(address));
Expand Down Expand Up @@ -107,6 +119,8 @@ function requestPinned(
resolved: ResolvedAddress,
maxBytes: number,
timeoutMs: number,
method: "GET" | "POST" = "GET",
requestBody = "",
): Promise<Omit<SafeHttpResponse, "finalUrl" | "redirects">> {
return new Promise((resolve, reject) => {
const request = (url.protocol === "https:" ? httpsRequest : httpRequest)({
Expand All @@ -115,11 +129,15 @@ function requestPinned(
family: resolved.family,
port: url.port ? Number(url.port) : undefined,
path: `${url.pathname}${url.search}`,
method: "GET",
method,
headers: {
Accept: "application/json, text/plain;q=0.9, */*;q=0.8",
Host: url.host,
"User-Agent": "AgentSentry-BusinessTool/1.0",
...(method === "POST" ? {
"Content-Type": "application/json; charset=utf-8",
"Content-Length": String(Buffer.byteLength(requestBody)),
} : {}),
},
servername: isIP(stripIpv6Brackets(url.hostname)) ? undefined : stripIpv6Brackets(url.hostname),
}, (response) => {
Expand Down Expand Up @@ -156,6 +174,7 @@ function requestPinned(
});
request.setTimeout(timeoutMs, () => request.destroy(new Error("HTTP request timed out")));
request.on("error", reject);
if (method === "POST" && requestBody) request.write(requestBody);
request.end();
});
}
Expand Down
Loading
Loading