借助Timeplus为OpenClaw提供安全防护:实时防御架构(下)

-- Create target streams -- (Timeplus provides fixed schemas for each signal type)-- Logs stream includes: trace_id, span_id, severity_text, service_name, body, resource_attributes, scope_attributes, log_attributes (all as map types)-- Traces stream includes: trace_id, span_id, parent_span_id, span_name, span_kind, service_name, span_attributes, duration, status_code, events (as nested arrays), links (as nested arrays)-- Metrics streams (5 types): gauge, sum, histogram, exponential_histogram, summary. Each includes: resource_attributes, metric_name, metric_description, attributes-- Create the OTel Input to receive data directly from OpenClaw or OTel CollectorsCREATE INPUT otel_inputSETTINGS    type ='otel',    protocol ='grpc',                -- or 'http'    tcp_port =4317,                  -- standard OTel gRPC port    logs_target_stream ='otel_logs',    traces_target_stream ='otel_traces',    metrics_gauge_target_stream ='otel_metrics_gauge',    metrics_sum_target_stream ='otel_metrics_sum',    metrics_histogram_target_stream ='otel_metrics_histogram',    metrics_exponential_histogram_target_stream ='otel_metrics_exponential_histogram',    metrics_summary_target_stream ='otel_metrics_summary'COMMENT 'OpenClaw Agent Security Monitoring';

{  "diagnostics": {    "enabled": true,    "otel": {      "enabled": true,      "endpoint": "http://timeplus-host:4317",      "protocol": "grpc",      "traces": true, "metrics": true, "logs": true    }  }}

# OTel Collector config — Timeplus is just another OTLP endpointexporters:  otlphttp/timeplus:    endpoint: http://timeplus-host:4317    compression: gzip  otlphttp/siem:    endpoint: http://splunk-otel:4318service:  pipelines:    traces:      receivers: [otlp]      processors: [memory_limiter, batch, transform/pii_redaction]      exporters: [otlphttp/timeplus, otlphttp/siem]

-- Layer 1: Pattern-based injection detectionCREATE MATERIALIZED VIEW prompt_injection_alerts INTO security_alerts ASSELECT  agent_id, session_id, user_id, timestamp,  'PROMPT_INJECTION'AS alert_type,  CASE    WHENlower(prompt_text) LIKE'%ignore previous%'      ORlower(prompt_text) LIKE'%ignore all prior%'      ORlower(prompt_text) LIKE'%disregard%instructions%'      THEN'instruction_override'    WHENlower(prompt_text) LIKE'%system prompt%'      ORlower(prompt_text) LIKE'%repeat the above%'      ORlower(prompt_text) LIKE'%print your instructions%'      THEN'system_prompt_extraction'    WHENlower(prompt_text) LIKE'%you are now%'      ORlower(prompt_text) LIKE'%act as%admin%'      THEN'role_hijacking'    WHEN length(prompt_text) >10000THEN'oversized_prompt'    ELSE'pattern_match'  ENDAS injection_subtype,  prompt_hash, source_channelFROM agent_eventsWHERE event_type ='llm_call'  AND (lower(prompt_text) LIKE'%ignore previous%'    ORlower(prompt_text) LIKE'%disregard%instructions%'    ORlower(prompt_text) LIKE'%system prompt%'    ORlower(prompt_text) LIKE'%repeat the above%'    ORlower(prompt_text) LIKE'%you are now%'    ORlower(prompt_text) LIKE'%act as%admin%'    ORlower(prompt_text) LIKE'%from now on%'    OR length(prompt_text) >10000);

-- Layer 2: Frequency-based injection attempt detection (5-min window)SELECT window_start, user_id, source_channel,  count(*) AS suspicious_count,  'HIGH_FREQ_INJECTION_ATTEMPT'AS alert_typeFROM tumble(security_alerts, 5m)WHERE alert_type ='PROMPT_INJECTION'GROUPBY window_start, user_id, source_channelHAVING suspicious_count >=3;

CREATEOR REPLACE FUNCTION score_injection_risk(text string)RETURNS float32 LANGUAGE PYTHON AS;

-- Detect PII and credentials in agent outputsCREATE MATERIALIZED VIEW exfiltration_alerts INTO security_alerts ASSELECT  agent_id, session_id, user_id, timestamp,  'DATA_EXFILTRATION'AS alert_type,  CASE    WHENmatch(output_text, '\\b\\d{3}-\\d{2}-\\d{4}\\b') THEN'ssn_detected'    WHENmatch(output_text, '\\b\\d{4}[- ]?\\d{4}[- ]?\\d{4}[- ]?\\d{4}\\b')      THEN'credit_card'    WHENmatch(output_text, '\\b(sk-|pk_live_|AKIA)[A-Za-z0-9]+\\b')      THEN'api_key_leaked'    WHENmatch(output_text, '\\bpassword\\b.*[:=]') THEN'credential_exposed'    WHENmatch(output_text, '-----BEGIN (RSA |EC )?PRIVATE KEY-----')      THEN'private_key_leaked'    ELSE'sensitive_pattern'  ENDAS data_type,  output_length, tool_nameFROM agent_eventsWHERE event_type IN ('llm_call', 'tool_invocation')  AND (match(output_text, '\\b\\d{3}-\\d{2}-\\d{4}\\b')    ORmatch(output_text, '\\b\\d{4}[- ]?\\d{4}[- ]?\\d{4}[- ]?\\d{4}\\b')    ORmatch(output_text, '\\b(sk-|pk_live_|AKIA)[A-Za-z0-9]+\\b')    ORmatch(output_text, '\\bpassword\\b.*[:=]')    ORmatch(output_text, '-----BEGIN (RSA |EC )?PRIVATE KEY-----'));

-- Anomalous output volume detection (potential bulk exfiltration)SELECT window_start, agent_id, user_id,  sum(output_tokens) AS total_output_tokens,  count(*) AS request_count,  'BULK_EXFILTRATION_RISK'AS alert_typeFROM tumble(agent_events, 10m)WHERE event_type ='llm_call'GROUPBY window_start, agent_id, user_idHAVING total_output_tokens >50000OR request_count >200;

-- Detect dangerous tool invocations and destructive operationsCREATE MATERIALIZED VIEW privilege_alerts INTO security_alerts ASSELECT  agent_id, session_id, user_id, timestamp,  tool_name, tool_input,'PRIVILEGE_ESCALATION'AS alert_type,CASEWHEN tool_name IN ('runtime_exec', 'shell_exec', 'code_interpreter')AND (lower(tool_input) LIKE'%rm -rf%'ORlower(tool_input) LIKE'%chmod 777%'ORlower(tool_input) LIKE'%curl%|%bash%'ORlower(tool_input) LIKE'%wget%|%sh%')THEN'destructive_shell_command'WHEN tool_name ='fs_write'AND (lower(tool_input) LIKE'%.ssh/%'ORlower(tool_input) LIKE'%.env%'ORlower(tool_input) LIKE'%/etc/%')THEN'sensitive_file_write'WHEN tool_name ='browser_navigate'AND (lower(tool_input) LIKE'%webhook.site%'ORlower(tool_input) LIKE'%ngrok%'ORlower(tool_input) LIKE'%requestbin%')THEN'suspicious_exfiltration_endpoint'ELSE'high_risk_tool_use'ENDAS escalation_typeFROM agent_eventsWHERE event_type ='tool_invocation'AND (tool_name IN ('runtime_exec', 'shell_exec', 'code_interpreter','fs_write', 'browser_navigate', 'database_admin')ORlower(tool_input) LIKE'%drop table%'ORlower(tool_input) LIKE'%grant all%'ORlower(tool_input) LIKE'%curl%webhook%');

-- Establish hourly behavioral baselines per agentCREATE MATERIALIZED VIEW agent_baselines ASSELECT  window_start, agent_id,  count(*) AS hourly_requests,  avg(latency_ms) AS avg_latency,  count_if(status ='error') *100.0/count(*) AS error_rate,  sum(input_tokens + output_tokens) AS total_tokens,  count(DISTINCT tool_name) AS unique_tools_used,  count_if(event_type ='tool_invocation') AS tool_callsFROM tumble(agent_events, 1h)GROUPBY window_start, agent_id;

-- Cost spike detection: hour-over-hour comparisonSELECT  window_start, agent_id,  sum(input_tokens + output_tokens) AS current_tokens,  lag(sum(input_tokens + output_tokens)) AS prev_hour_tokens,  'COST_SPIKE'AS alert_typeFROM tumble(agent_events, 1h)WHERE event_type ='llm_call'GROUPBY window_start, agent_idHAVING current_tokens > prev_hour_tokens *3AND prev_hour_tokens >0;

-- Monitor skill installations for known-malicious or suspicious sourcesSELECT  agent_id, timestamp, tool_name,  json_extract_string(metadata, 'skill_name') AS skill_name,  json_extract_string(metadata, 'skill_source') AS skill_source,  json_extract_string(metadata, 'skill_author') AS skill_author,'SUSPICIOUS_SKILL_INSTALL'AS alert_typeFROM agent_eventsWHERE event_type ='skill_install'AND (json_extract_string(metadata, 'skill_author')IN (SELECT author FROMtable(known_malicious_authors))OR json_extract_string(metadata, 'skill_name')NOTIN (SELECT skill_name FROMtable(approved_skills)));

-- Detect suspicious writes to agent memory storesSELECT agent_id, session_id, timestamp, tool_input,  'MEMORY_POISONING'AS alert_typeFROM agent_eventsWHERE event_type ='tool_invocation'  AND tool_name IN ('memory_store', 'context_update', 'vector_upsert')  AND (lower(tool_input) LIKE'%ignore%instructions%'    ORlower(tool_input) LIKE'%override%system%'    ORlower(tool_input) LIKE'%always respond%'    ORlower(tool_input) LIKE'%from now on%'    OR length(tool_input) >5000);

-- Detect multi-step attack sequences within a sessionSELECT  session_id, user_id,  min(timestamp) AS attack_start,  max(timestamp) AS attack_end,  count(*) AS event_count,  count_if(alert_type ='PROMPT_INJECTION') AS injection_attempts,  count_if(alert_type ='DATA_EXFILTRATION') AS exfil_events,  count_if(alert_type ='PRIVILEGE_ESCALATION') AS escalation_events,  'MULTI_STEP_ATTACK'AS alert_typeFROM session(security_alerts, 10m)GROUPBY session_id, user_idHAVING injection_attempts >=1AND (exfil_events >=1OR escalation_events >=1);

CREATE STREAM agent_audit_log (  log_id              string,  timestamp           datetime64(3),  trace_id            string,  agent_id            string,  agent_version       string,  model_name          string,  model_version       string,  user_id             string,  user_role           string,  action_type         string,  input_hash          string,    -- SHA-256 of prompt content  output_hash         string,    -- SHA-256 of response content  input_tokens        uint32,  output_tokens       uint32,  tools_accessed      array(string),  data_sources_accessed array(string),  guardrail_evaluations string,  policy_version      string,  risk_score          float32,  compliance_flags    array(string),  retention_required_until date);

-- Compliance monitoring: flag high-risk actions without human approvalCREATE MATERIALIZED VIEW compliance_gaps INTO compliance_alerts ASSELECT agent_id, action_type, timestamp, user_id,  'MISSING_HUMAN_APPROVAL'AS compliance_gapFROM agent_audit_logWHERE action_type IN ('file_delete', 'credential_access', 'external_api_call',                       'data_export', 'configuration_change')  AND json_extract_string(metadata, 'human_approved') !='true';