assay_core/mcp/
proxy.rs

1use super::audit::{AuditEvent, AuditLog};
2use super::decision::{
3    reason_codes, refresh_contract_projections, Decision, DecisionEmitter, DecisionEvent,
4    FileDecisionEmitter, NullDecisionEmitter,
5};
6use super::jsonrpc::JsonRpcRequest;
7use super::policy::{
8    make_deny_response, McpPolicy, PolicyDecision, PolicyMatchMetadata, PolicyState,
9};
10use super::tool_definition::{binding_from_tools_list_tool, ToolDefinitionBinding};
11use std::{
12    collections::HashMap,
13    io::{self, BufRead, BufReader, Write},
14    process::{Child, Command, Stdio},
15    sync::{Arc, Mutex},
16    thread,
17};
18
19/// Validated proxy configuration.
20///
21/// Use `ProxyConfig::try_from_raw()` to create from CLI/config input.
22#[derive(Clone, Debug)]
23pub struct ProxyConfig {
24    pub dry_run: bool,
25    pub verbose: bool,
26    /// NDJSON log for mandate lifecycle events (audit trail)
27    pub audit_log_path: Option<std::path::PathBuf>,
28    pub server_id: String,
29    /// NDJSON log for tool decision events (high volume)
30    pub decision_log_path: Option<std::path::PathBuf>,
31    /// CloudEvents source URI (validated, required when logging enabled)
32    pub event_source: Option<String>,
33}
34
35/// Raw config as provided by CLI/config files before validation.
36#[derive(Clone, Debug, Default)]
37pub struct ProxyConfigRaw {
38    pub dry_run: bool,
39    pub verbose: bool,
40    pub audit_log_path: Option<std::path::PathBuf>,
41    pub server_id: String,
42    pub decision_log_path: Option<std::path::PathBuf>,
43    pub event_source: Option<String>,
44}
45
46impl ProxyConfig {
47    /// Create validated config from raw input.
48    ///
49    /// Fails if:
50    /// - Logging is enabled but event_source is missing
51    /// - event_source is not a valid absolute URI (scheme://...)
52    pub fn try_from_raw(raw: ProxyConfigRaw) -> anyhow::Result<Self> {
53        let logging_enabled = raw.audit_log_path.is_some() || raw.decision_log_path.is_some();
54
55        let event_source = raw
56            .event_source
57            .map(|s| s.trim().to_string())
58            .filter(|s| !s.is_empty());
59
60        if logging_enabled && event_source.is_none() {
61            anyhow::bail!(
62                "event_source is required when logging is enabled (e.g. --event-source assay://org/app)"
63            );
64        }
65
66        if let Some(ref src) = event_source {
67            validate_event_source(src)?;
68        }
69
70        Ok(ProxyConfig {
71            dry_run: raw.dry_run,
72            verbose: raw.verbose,
73            audit_log_path: raw.audit_log_path,
74            server_id: raw.server_id,
75            decision_log_path: raw.decision_log_path,
76            event_source,
77        })
78    }
79}
80
81/// Validate event_source URI (must be absolute with scheme://).
82fn validate_event_source(s: &str) -> anyhow::Result<()> {
83    let s = s.trim();
84    if s.is_empty() {
85        anyhow::bail!("event_source must be absolute URI with scheme (e.g. assay://org/app)");
86    }
87    if s.chars().any(|c| c.is_whitespace()) {
88        anyhow::bail!("event_source must not contain whitespace");
89    }
90
91    // Require scheme://...
92    let Some(pos) = s.find("://") else {
93        anyhow::bail!("event_source must be absolute URI with scheme (e.g. assay://org/app)");
94    };
95    if pos == 0 {
96        anyhow::bail!("event_source must have scheme before :// (e.g. assay://org/app)");
97    }
98
99    // Validate scheme charset (RFC 3986: ALPHA *( ALPHA / DIGIT / "+" / "-" / "." ))
100    let scheme = &s[..pos];
101    let mut chars = scheme.chars();
102    match chars.next() {
103        Some(c) if c.is_ascii_alphabetic() => {}
104        _ => anyhow::bail!("event_source URI scheme must start with a letter"),
105    }
106    if !chars.all(|c| c.is_ascii_alphanumeric() || c == '+' || c == '-' || c == '.') {
107        anyhow::bail!("event_source URI scheme contains invalid characters");
108    }
109
110    Ok(())
111}
112
113pub struct McpProxy {
114    child: Child,
115    policy: McpPolicy,
116    config: ProxyConfig,
117    /// Cache of tool identities discovered during tools/list
118    identity_cache: Arc<Mutex<HashMap<String, super::identity::ToolIdentity>>>,
119    /// Cache of bounded tool-definition bindings discovered during tools/list
120    tool_definition_cache: Arc<Mutex<HashMap<String, ToolDefinitionBinding>>>,
121}
122
123impl Drop for McpProxy {
124    fn drop(&mut self) {
125        // Best-effort cleanup
126        let _ = self.child.kill();
127    }
128}
129
130impl McpProxy {
131    pub fn spawn(
132        command: &str,
133        args: &[String],
134        policy: McpPolicy,
135        config: ProxyConfig,
136    ) -> io::Result<Self> {
137        let child = Command::new(command)
138            .args(args)
139            .stdin(Stdio::piped())
140            .stdout(Stdio::piped())
141            .stderr(Stdio::inherit()) // protocol blijft op stdout
142            .spawn()?;
143
144        Ok(Self {
145            child,
146            policy,
147            config,
148            identity_cache: Arc::new(Mutex::new(HashMap::new())),
149            tool_definition_cache: Arc::new(Mutex::new(HashMap::new())),
150        })
151    }
152
153    pub fn run(mut self) -> io::Result<i32> {
154        let mut child_stdin = self.child.stdin.take().expect("child stdin");
155        let child_stdout = self.child.stdout.take().expect("child stdout");
156
157        let stdout = Arc::new(Mutex::new(io::stdout()));
158        let policy = self.policy.clone();
159        let config = self.config.clone();
160        let identity_cache_a = self.identity_cache.clone();
161        let identity_cache_b = self.identity_cache.clone();
162        let tool_definition_cache_a = self.tool_definition_cache.clone();
163        let tool_definition_cache_b = self.tool_definition_cache.clone();
164
165        // Initialize decision emitter (I1: always emit decision)
166        let decision_emitter: Arc<dyn DecisionEmitter> =
167            if let Some(path) = &config.decision_log_path {
168                Arc::new(FileDecisionEmitter::new(path)?)
169            } else {
170                Arc::new(NullDecisionEmitter)
171            };
172        let event_source = config
173            .event_source
174            .clone()
175            .unwrap_or_else(|| format!("assay://{}", config.server_id));
176
177        // Thread A: server -> client passthrough
178        let stdout_a = stdout.clone();
179        let t_server_to_client = thread::spawn(move || -> io::Result<()> {
180            let mut reader = BufReader::new(child_stdout);
181            let mut line = String::new();
182
183            while reader.read_line(&mut line)? > 0 {
184                let mut processed_line = line.clone();
185
186                // Phase 9: Compute Identities on tools/list response
187                if let Ok(mut v) = serde_json::from_str::<serde_json::Value>(&line) {
188                    if let Some(result) = v.get_mut("result") {
189                        if let Some(tools) = result.get_mut("tools").and_then(|t| t.as_array_mut())
190                        {
191                            for tool in tools {
192                                if let Some(observation) =
193                                    Self::observe_tool_definition(tool, &config.server_id)
194                                {
195                                    let mut identity_cache = identity_cache_a.lock().unwrap();
196                                    identity_cache.insert(
197                                        observation.name.clone(),
198                                        observation.identity.clone(),
199                                    );
200                                    drop(identity_cache);
201
202                                    if let Some(binding) = observation.binding {
203                                        let mut binding_cache =
204                                            tool_definition_cache_a.lock().unwrap();
205                                        binding_cache.insert(observation.name, binding);
206                                    }
207                                }
208                            }
209                            processed_line =
210                                serde_json::to_string(&v).unwrap_or(line.clone()) + "\n";
211                        }
212                    }
213                }
214
215                let mut out = stdout_a
216                    .lock()
217                    .map_err(|e| io::Error::other(e.to_string()))?;
218                out.write_all(processed_line.as_bytes())?;
219                out.flush()?;
220                line.clear();
221            }
222            Ok(())
223        });
224
225        // Thread B: client -> server passthrough with Policy Check
226        let stdout_b = stdout.clone();
227        let emitter_b = decision_emitter.clone();
228        let event_source_b = event_source.clone();
229        let t_client_to_server = thread::spawn(move || -> io::Result<()> {
230            let stdin = io::stdin();
231            let mut reader = stdin.lock();
232            let mut line = String::new();
233
234            let mut state = PolicyState::default();
235            let mut audit_log = AuditLog::new(config.audit_log_path.as_deref());
236
237            while reader.read_line(&mut line)? > 0 {
238                // 1. Try Parse as MCP Request
239                match serde_json::from_str::<JsonRpcRequest>(&line) {
240                    Ok(req) => {
241                        // 2. Check Policy with Identity (Phase 9)
242                        let (runtime_id, tool_definition_binding) = if req.is_tool_call() {
243                            let name = req.tool_params().map(|p| p.name).unwrap_or_default();
244                            let runtime_id = {
245                                let cache = identity_cache_b.lock().unwrap();
246                                cache.get(&name).cloned()
247                            };
248                            let tool_definition_binding = {
249                                let cache = tool_definition_cache_b.lock().unwrap();
250                                cache.get(&name).cloned()
251                            };
252                            (runtime_id, tool_definition_binding)
253                        } else {
254                            (None, None)
255                        };
256
257                        let tool_name = req.tool_params().map(|p| p.name).unwrap_or_default();
258                        let tool_call_id = Self::extract_tool_call_id(&req);
259
260                        let policy_eval = policy.evaluate_with_metadata(
261                            &tool_name,
262                            &req.tool_params()
263                                .map(|p| p.arguments)
264                                .unwrap_or(serde_json::Value::Null),
265                            &mut state,
266                            runtime_id.as_ref(),
267                        );
268
269                        match policy_eval.decision {
270                            PolicyDecision::Allow => {
271                                Self::handle_allow(&req, &mut audit_log, config.verbose);
272                                // Emit decision event (I1: always emit)
273                                if req.is_tool_call() {
274                                    Self::emit_decision(
275                                        &emitter_b,
276                                        &event_source_b,
277                                        &tool_call_id,
278                                        &tool_name,
279                                        Decision::Allow,
280                                        reason_codes::P_POLICY_ALLOW,
281                                        None,
282                                        req.id.clone(),
283                                        &policy_eval.metadata,
284                                        tool_definition_binding.as_ref(),
285                                    );
286                                }
287                            }
288                            PolicyDecision::AllowWithWarning { tool, code, reason } => {
289                                // Log warning about allowing a tool invocation with issues
290                                if config.verbose {
291                                    eprintln!(
292                                        "[assay] WARNING: Allowing tool '{}' with warning (code: {}, reason: {}).",
293                                        tool,
294                                        code,
295                                        reason
296                                    );
297                                }
298                                audit_log.log(&AuditEvent {
299                                    timestamp: chrono::Utc::now().to_rfc3339(),
300                                    decision: "allow_with_warning".to_string(),
301                                    tool: Some(tool.clone()),
302                                    reason: Some(reason.clone()),
303                                    request_id: req.id.clone(),
304                                    agentic: None,
305                                });
306                                // Emit decision event (I1: always emit)
307                                Self::emit_decision(
308                                    &emitter_b,
309                                    &event_source_b,
310                                    &tool_call_id,
311                                    &tool,
312                                    Decision::Allow,
313                                    &code,
314                                    Some(reason),
315                                    req.id.clone(),
316                                    &policy_eval.metadata,
317                                    tool_definition_binding.as_ref(),
318                                );
319                                // Then proceed as a normal allow
320                                Self::handle_allow(&req, &mut audit_log, false);
321                                // false = don't double log ALLOW
322                            }
323                            PolicyDecision::Deny {
324                                tool,
325                                code,
326                                reason,
327                                contract,
328                            } => {
329                                // Log Decision
330                                let decision_str =
331                                    if config.dry_run { "would_deny" } else { "deny" };
332
333                                if config.verbose {
334                                    eprintln!(
335                                        "[assay] {} {} (reason: {})",
336                                        decision_str.to_uppercase(),
337                                        tool,
338                                        reason
339                                    );
340                                }
341
342                                audit_log.log(&AuditEvent {
343                                    timestamp: chrono::Utc::now().to_rfc3339(),
344                                    decision: decision_str.to_string(),
345                                    tool: Some(tool.clone()),
346                                    reason: Some(reason.clone()),
347                                    request_id: req.id.clone(),
348                                    agentic: Some(contract.clone()),
349                                });
350
351                                // Emit decision event (I1: always emit)
352                                let reason_code = Self::map_policy_code(&code);
353                                Self::emit_decision(
354                                    &emitter_b,
355                                    &event_source_b,
356                                    &tool_call_id,
357                                    &tool,
358                                    if config.dry_run {
359                                        Decision::Allow
360                                    } else {
361                                        Decision::Deny
362                                    },
363                                    &reason_code,
364                                    Some(reason),
365                                    req.id.clone(),
366                                    &policy_eval.metadata,
367                                    tool_definition_binding.as_ref(),
368                                );
369
370                                if config.dry_run {
371                                    // DRY RUN: Forward anyway
372                                    // Fallthrough to forward logic below
373                                } else {
374                                    // BLOCK: Send error response
375                                    let id = req.id.unwrap_or(serde_json::Value::Null);
376                                    let response_json = make_deny_response(
377                                        id,
378                                        "Content blocked by policy",
379                                        contract,
380                                    );
381
382                                    let mut out = stdout_b
383                                        .lock()
384                                        .map_err(|e| io::Error::other(e.to_string()))?;
385                                    out.write_all(response_json.as_bytes())?;
386                                    out.flush()?;
387
388                                    line.clear();
389                                    continue; // Skip forwarding
390                                }
391                            }
392                        }
393                    }
394                    Err(_) => {
395                        // Hardening: Suspicious Unparsable JSON
396                        let trimmed = line.trim();
397                        if trimmed.starts_with('{')
398                            && (trimmed.contains("\"method\"")
399                                || trimmed.contains("\"params\"")
400                                || trimmed.contains("\"tool\""))
401                        {
402                            eprintln!("[assay] WARNING: Suspicious unparsable JSON, forwarding anyway (potential bypass attempt?): {:.60}...", trimmed);
403                        }
404                    }
405                }
406
407                // 3. Forward
408                child_stdin.write_all(line.as_bytes())?;
409                child_stdin.flush()?;
410                line.clear();
411            }
412            Ok(())
413        });
414
415        // Wacht tot client->server eindigt (stdin closed)
416        t_client_to_server
417            .join()
418            .map_err(|_| io::Error::other("client->server thread panicked"))??;
419
420        // Server->client thread kan nog even lopen; join best-effort
421        let _ = t_server_to_client.join();
422
423        // Wacht op child exit
424        let status = self.child.wait()?;
425        Ok(status.code().unwrap_or(1))
426    }
427
428    fn handle_allow(req: &JsonRpcRequest, audit_log: &mut AuditLog, verbose: bool) {
429        if verbose && req.is_tool_call() {
430            let tool = req
431                .tool_params()
432                .map(|p| p.name)
433                .unwrap_or_else(|| "unknown".to_string());
434            eprintln!("[assay] ALLOW {}", tool);
435        }
436
437        if req.is_tool_call() {
438            let tool = req.tool_params().map(|p| p.name);
439            audit_log.log(&AuditEvent {
440                timestamp: chrono::Utc::now().to_rfc3339(),
441                decision: "allow".to_string(),
442                tool,
443                reason: None,
444                request_id: req.id.clone(),
445                agentic: None,
446            });
447        }
448    }
449
450    /// Extract tool_call_id from request (I4: idempotency key).
451    fn extract_tool_call_id(request: &JsonRpcRequest) -> String {
452        // Try to get from params._meta.tool_call_id (MCP standard)
453        if let Some(params) = request.tool_params() {
454            if let Some(meta) = params.arguments.get("_meta") {
455                if let Some(id) = meta.get("tool_call_id").and_then(|v| v.as_str()) {
456                    return id.to_string();
457                }
458            }
459        }
460
461        // Fall back to request.id if present
462        if let Some(id) = &request.id {
463            if let Some(s) = id.as_str() {
464                return format!("req_{}", s);
465            }
466            if let Some(n) = id.as_i64() {
467                return format!("req_{}", n);
468            }
469        }
470
471        // Generate one if none found
472        format!("gen_{}", uuid::Uuid::new_v4())
473    }
474
475    /// Map policy error code to reason code.
476    fn map_policy_code(code: &str) -> String {
477        match code {
478            "E_TOOL_DENIED" => reason_codes::P_TOOL_DENIED.to_string(),
479            "E_TOOL_NOT_ALLOWED" => reason_codes::P_TOOL_NOT_ALLOWED.to_string(),
480            "E_ARG_SCHEMA" => reason_codes::P_ARG_SCHEMA.to_string(),
481            "E_RATE_LIMIT" => reason_codes::P_RATE_LIMIT.to_string(),
482            "E_TOOL_DRIFT" => reason_codes::P_TOOL_DRIFT.to_string(),
483            _ => reason_codes::P_POLICY_DENY.to_string(),
484        }
485    }
486
487    /// Emit a decision event (I1: always emit).
488    #[allow(clippy::too_many_arguments)]
489    fn emit_decision(
490        emitter: &Arc<dyn DecisionEmitter>,
491        source: &str,
492        tool_call_id: &str,
493        tool: &str,
494        decision: Decision,
495        reason_code: &str,
496        reason: Option<String>,
497        request_id: Option<serde_json::Value>,
498        metadata: &PolicyMatchMetadata,
499        tool_definition_binding: Option<&ToolDefinitionBinding>,
500    ) {
501        let mut event = DecisionEvent::new(
502            source.to_string(),
503            tool_call_id.to_string(),
504            tool.to_string(),
505        );
506        event.data.decision = decision;
507        event.data.reason_code = reason_code.to_string();
508        event.data.reason = reason;
509        event.data.request_id = request_id;
510        event.data.tool_classes = metadata.tool_classes.clone();
511        event.data.matched_tool_classes = metadata.matched_tool_classes.clone();
512        event.data.match_basis = metadata.match_basis.as_str().map(ToString::to_string);
513        event.data.matched_rule = metadata.matched_rule.clone();
514        event.data.typed_decision = metadata.typed_decision;
515        event.data.policy_version = metadata.policy_version.clone();
516        event.data.policy_digest = metadata.policy_digest.clone();
517        event.data.apply_policy_snapshot_projection();
518        event
519            .data
520            .apply_tool_definition_binding(tool_definition_binding);
521        event.data.obligations = metadata.obligations.clone();
522        event.data.obligation_outcomes =
523            super::obligations::execute_log_only(&metadata.obligations, tool);
524        event.data.approval_state = metadata.approval_state.clone();
525        if let Some(artifact) = &metadata.approval_artifact {
526            event.data.approval_id = Some(artifact.approval_id.clone());
527            event.data.approver = Some(artifact.approver.clone());
528            event.data.issued_at = Some(artifact.issued_at.clone());
529            event.data.expires_at = Some(artifact.expires_at.clone());
530            event.data.scope = Some(artifact.scope.clone());
531            event.data.approval_bound_tool = Some(artifact.bound_tool.clone());
532            event.data.approval_bound_resource = Some(artifact.bound_resource.clone());
533        }
534        event.data.approval_freshness = metadata.approval_freshness;
535        event.data.approval_failure_reason = metadata.approval_failure_reason.clone();
536        event.data.scope_type = metadata.scope_type.clone();
537        event.data.scope_value = metadata.scope_value.clone();
538        event.data.scope_match_mode = metadata.scope_match_mode.clone();
539        event.data.scope_evaluation_state = metadata.scope_evaluation_state.clone();
540        event.data.scope_failure_reason = metadata.scope_failure_reason.clone();
541        event.data.restrict_scope_present = metadata.restrict_scope_present;
542        event.data.restrict_scope_target = metadata.restrict_scope_target.clone();
543        event.data.restrict_scope_match = metadata.restrict_scope_match;
544        event.data.restrict_scope_reason = metadata.restrict_scope_reason.clone();
545        event.data.redaction_target = metadata.redaction_target.clone();
546        event.data.redaction_mode = metadata.redaction_mode.clone();
547        event.data.redaction_scope = metadata.redaction_scope.clone();
548        event.data.redaction_applied_state = metadata.redaction_applied_state.clone();
549        event.data.redaction_reason = metadata.redaction_reason.clone();
550        event.data.redaction_failure_reason = metadata.redaction_failure_reason.clone();
551        event.data.redact_args_present = metadata.redact_args_present;
552        event.data.redact_args_target = metadata.redact_args_target.clone();
553        event.data.redact_args_mode = metadata.redact_args_mode.clone();
554        event.data.redact_args_result = metadata.redact_args_result.clone();
555        event.data.redact_args_reason = metadata.redact_args_reason.clone();
556        event.data.fail_closed = metadata.fail_closed.clone();
557        event.data.lane = metadata.lane.clone();
558        event.data.principal = metadata.principal.clone();
559        event.data.auth_context_summary = metadata.auth_context_summary.clone();
560        event.data.auth_scheme = metadata.auth_scheme.clone();
561        event.data.auth_issuer = metadata.auth_issuer.clone();
562        event.data.delegated_from = metadata.delegated_from.clone();
563        event.data.delegation_depth = metadata.delegation_depth;
564        refresh_contract_projections(&mut event.data);
565        emitter.emit(&event);
566    }
567
568    fn observe_tool_definition(
569        tool: &mut serde_json::Value,
570        server_id: &str,
571    ) -> Option<ToolDefinitionObservation> {
572        let name = tool.get("name").and_then(|n| n.as_str())?;
573        if name.trim().is_empty() {
574            return None;
575        }
576        let name = name.to_string();
577        let description = tool
578            .get("description")
579            .and_then(|d| d.as_str())
580            .map(|s| s.to_string());
581        let input_schema = tool
582            .get("inputSchema")
583            .or_else(|| tool.get("input_schema"))
584            .cloned();
585
586        let identity =
587            super::identity::ToolIdentity::new(server_id, &name, &input_schema, &description);
588        let binding = binding_from_tools_list_tool(tool, Some(server_id))
589            .ok()
590            .flatten();
591
592        // Augment the response with the computed identity for downstream/logging.
593        tool.as_object_mut().and_then(|m| {
594            m.insert(
595                "tool_identity".to_string(),
596                serde_json::to_value(&identity).unwrap(),
597            )
598        });
599
600        Some(ToolDefinitionObservation {
601            name,
602            identity,
603            binding,
604        })
605    }
606}
607
608struct ToolDefinitionObservation {
609    name: String,
610    identity: super::identity::ToolIdentity,
611    binding: Option<ToolDefinitionBinding>,
612}
613
614#[cfg(test)]
615mod tests {
616    use super::*;
617    use crate::mcp::tool_definition::{
618        TOOL_DEFINITION_CANONICALIZATION_JCS_MCP_TOOL_DEFINITION_V1,
619        TOOL_DEFINITION_DIGEST_ALG_SHA256, TOOL_DEFINITION_SCHEMA_V1,
620        TOOL_DEFINITION_SOURCE_MCP_TOOLS_LIST,
621    };
622    use std::sync::Mutex as StdMutex;
623
624    struct CapturingEmitter {
625        events: StdMutex<Vec<DecisionEvent>>,
626    }
627
628    impl CapturingEmitter {
629        fn new() -> Self {
630            Self {
631                events: StdMutex::new(Vec::new()),
632            }
633        }
634    }
635
636    impl DecisionEmitter for CapturingEmitter {
637        fn emit(&self, event: &DecisionEvent) {
638            self.events.lock().unwrap().push(event.clone());
639        }
640    }
641
642    #[test]
643    fn event_source_accepts_assay_uri() {
644        validate_event_source("assay://myorg/myapp").unwrap();
645    }
646
647    #[test]
648    fn event_source_accepts_https_uri() {
649        validate_event_source("https://example.com/agent").unwrap();
650    }
651
652    #[test]
653    fn event_source_rejects_empty() {
654        assert!(validate_event_source("").is_err());
655        assert!(validate_event_source("   ").is_err());
656    }
657
658    #[test]
659    fn event_source_rejects_whitespace() {
660        assert!(validate_event_source("assay://myorg/my app").is_err());
661        assert!(validate_event_source("assay://myorg/\tmyapp").is_err());
662    }
663
664    #[test]
665    fn event_source_rejects_missing_scheme() {
666        assert!(validate_event_source("myorg/myapp").is_err());
667        assert!(validate_event_source("://myorg/myapp").is_err());
668    }
669
670    #[test]
671    fn event_source_rejects_did_and_urn() {
672        // We require scheme:// not just scheme:
673        assert!(validate_event_source("did:example:123").is_err());
674        assert!(validate_event_source("urn:example:foo").is_err());
675    }
676
677    #[test]
678    fn event_source_rejects_scheme_starting_with_non_letter() {
679        assert!(validate_event_source("1assay://myorg/myapp").is_err());
680        assert!(validate_event_source("-assay://myorg/myapp").is_err());
681    }
682
683    #[test]
684    fn event_source_rejects_scheme_with_invalid_chars() {
685        assert!(validate_event_source("as_say://myorg/myapp").is_err());
686        assert!(validate_event_source("as@say://myorg/myapp").is_err());
687    }
688
689    #[test]
690    fn config_requires_event_source_when_logging_enabled() {
691        let raw = ProxyConfigRaw {
692            dry_run: false,
693            verbose: false,
694            audit_log_path: None,
695            decision_log_path: Some(std::path::PathBuf::from("decisions.ndjson")),
696            event_source: None,
697            server_id: "srv".to_string(),
698        };
699
700        let err = ProxyConfig::try_from_raw(raw).unwrap_err();
701        let msg = format!("{err:#}");
702        assert!(msg.contains("event_source is required"));
703    }
704
705    #[test]
706    fn config_allows_no_event_source_when_logging_disabled() {
707        let raw = ProxyConfigRaw {
708            dry_run: false,
709            verbose: false,
710            audit_log_path: None,
711            decision_log_path: None,
712            event_source: None,
713            server_id: "srv".to_string(),
714        };
715
716        ProxyConfig::try_from_raw(raw).unwrap();
717    }
718
719    #[test]
720    fn config_accepts_valid_event_source() {
721        let raw = ProxyConfigRaw {
722            dry_run: false,
723            verbose: false,
724            audit_log_path: None,
725            decision_log_path: Some(std::path::PathBuf::from("decisions.ndjson")),
726            event_source: Some("assay://myorg/myapp".to_string()),
727            server_id: "srv".to_string(),
728        };
729
730        let cfg = ProxyConfig::try_from_raw(raw).unwrap();
731        assert_eq!(cfg.event_source.as_deref(), Some("assay://myorg/myapp"));
732    }
733
734    #[test]
735    fn config_rejects_invalid_event_source_uri() {
736        let raw = ProxyConfigRaw {
737            dry_run: false,
738            verbose: false,
739            audit_log_path: None,
740            decision_log_path: Some(std::path::PathBuf::from("decisions.ndjson")),
741            event_source: Some("not a uri".to_string()),
742            server_id: "srv".to_string(),
743        };
744
745        assert!(ProxyConfig::try_from_raw(raw).is_err());
746    }
747
748    #[test]
749    fn observe_tool_definition_computes_identity_and_binding() {
750        let mut tool = serde_json::json!({
751            "name": "read_file",
752            "description": " Read files ",
753            "inputSchema": {"type": "object"},
754            "annotations": {"title": "Read"},
755            "x-assay-sig": {"signature": "opaque"}
756        });
757
758        let observation = McpProxy::observe_tool_definition(&mut tool, "server-a")
759            .expect("supported tool definition should be observed");
760
761        assert_eq!(observation.name, "read_file");
762        assert_eq!(observation.identity.server_id, "server-a");
763        assert!(observation.binding.is_some());
764        assert!(tool.get("tool_identity").is_some());
765    }
766
767    #[test]
768    fn emit_decision_projects_tool_definition_binding_atomically() {
769        let mut tool = serde_json::json!({
770            "name": "read_file",
771            "description": "Read files",
772            "inputSchema": {"type": "object"}
773        });
774        let observation = McpProxy::observe_tool_definition(&mut tool, "server-a")
775            .expect("supported tool definition should be observed");
776        let binding = observation.binding.expect("binding should be visible");
777        let emitter = Arc::new(CapturingEmitter::new());
778        let emitter_trait: Arc<dyn DecisionEmitter> = emitter.clone();
779
780        McpProxy::emit_decision(
781            &emitter_trait,
782            "assay://test",
783            "tc_tool_definition",
784            "read_file",
785            Decision::Allow,
786            reason_codes::P_POLICY_ALLOW,
787            None,
788            None,
789            &PolicyMatchMetadata::default(),
790            Some(&binding),
791        );
792
793        let events = emitter.events.lock().unwrap();
794        let data = &events[0].data;
795        assert!(data.tool_definition_digest.is_some());
796        assert_eq!(
797            data.tool_definition_digest_alg.as_deref(),
798            Some(TOOL_DEFINITION_DIGEST_ALG_SHA256)
799        );
800        assert_eq!(
801            data.tool_definition_canonicalization.as_deref(),
802            Some(TOOL_DEFINITION_CANONICALIZATION_JCS_MCP_TOOL_DEFINITION_V1)
803        );
804        assert_eq!(
805            data.tool_definition_schema.as_deref(),
806            Some(TOOL_DEFINITION_SCHEMA_V1)
807        );
808        assert_eq!(
809            data.tool_definition_source.as_deref(),
810            Some(TOOL_DEFINITION_SOURCE_MCP_TOOLS_LIST)
811        );
812    }
813}
assay_core/mcp/proxy.rs

assay_core/mcp/
proxy.rs
