From b9db2afd466006029d80732247c1c021b3ecc68a Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 10 Sep 2026 13:29:47 +0200 Subject: [PATCH 1/3] Python: tighten security label enforcement Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../packages/core/agent_framework/security.py | 481 ++++++++-- python/packages/core/tests/test_security.py | 832 +++++++++++++++++- .../tests/hosting_a2a/test_conversion.py | 33 + 3 files changed, 1267 insertions(+), 79 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index a655a9971af..85a2cf07242 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -44,6 +44,7 @@ from ._mcp import MCPTool __all__ = [ + "PRINCIPAL_METADATA_KEY", "SECURITY_TOOL_INSTRUCTIONS", "ConfidentialityLabel", "ContentLabel", @@ -76,9 +77,12 @@ _WHOLE_VAR_REF_RE = re.compile(rf"\s*(?:\[\s*(?P{_BRACKETED_VAR_ID})\s*\]|(?P{_BARE_VAR_ID}))\s*") _BARE_REFERENCE_WARNING = "Expanded a bare variable reference in a tool argument; models should use [var_] instead." _UNRESOLVED = object() -_AUTHORITATIVE_CONFIDENTIALITY = "_security_label_authoritative_confidentiality" +_AUTHORITATIVE_SECURITY_LABEL = "_security_label_authoritative" _INSPECT_VARIABLE_ERROR = "_inspect_variable_error" _INTERNAL_RESULT_MARKER = object() +_MAX_VARIABLE_REFERENCE_DEPTH = 16 +_MAX_VARIABLE_REFERENCE_COUNT = 100 +PRINCIPAL_METADATA_KEY = "agent_framework.security.principals" # Tools that consume variable IDs literally (as opaque references) and therefore # must NOT have ``var_xxx`` arguments expanded to stored content before execution. @@ -94,7 +98,40 @@ def _get_additional_properties(obj: Any) -> dict[str, Any]: return cast(dict[str, Any], props) if isinstance(props, dict) else {} -def _parse_content_label(label_data: MutableMapping[str, Any], *, source: str) -> ContentLabel: +def _canonical_principals(value: Any, *, source: str) -> tuple[tuple[str, str], ...]: + """Validate and canonicalize a principal-set declaration.""" + if not isinstance(value, list) or not value: + raise ValueError(f"{source} principals must be a non-empty list") + + principals: set[tuple[str, str]] = set() + for item in cast(list[Any], value): + if not isinstance(item, dict): + raise ValueError(f"{source} principals must contain mappings") + principal = cast(dict[str, Any], item) + if set(principal) != {"tenant_id", "user_id"}: + raise ValueError(f"{source} principal fields must be tenant_id and user_id") + tenant_id = principal.get("tenant_id") + user_id = principal.get("user_id") + if type(tenant_id) is not str or not tenant_id.strip() or type(user_id) is not str or not user_id.strip(): + raise ValueError(f"{source} principal identifiers must be non-empty strings") + principals.add((tenant_id, user_id)) + return tuple(sorted(principals)) + + +def _principal_list(principals: Sequence[tuple[str, str]]) -> list[dict[str, str]]: + """Return the serialized canonical representation of a principal set.""" + return [{"tenant_id": tenant_id, "user_id": user_id} for tenant_id, user_id in principals] + + +def _principal_binding_key(value: Any, *, source: str) -> str: + """Return a stable approval-binding fragment for a valid principal set.""" + principals = _principal_list(_canonical_principals(value, source=source)) + return json.dumps(principals, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _parse_content_label( + label_data: MutableMapping[str, Any], *, source: str, allow_principals: bool = False +) -> ContentLabel: """Parse explicit label fields while ignoring malformed optional metadata.""" integrity = label_data.get("integrity") confidentiality = label_data.get("confidentiality") @@ -105,10 +142,19 @@ def _parse_content_label(label_data: MutableMapping[str, Any], *, source: str) - if metadata is not None and not isinstance(metadata, dict): logger.warning("Ignoring malformed metadata from %s security label", source) metadata = {} + parsed_metadata = dict(cast(dict[str, Any], metadata)) if isinstance(metadata, dict) else {} + if not allow_principals: + parsed_metadata.pop(PRINCIPAL_METADATA_KEY, None) + elif confidentiality == ConfidentialityLabel.USER_IDENTITY.value: + parsed_metadata[PRINCIPAL_METADATA_KEY] = _principal_list( + _canonical_principals(parsed_metadata.get(PRINCIPAL_METADATA_KEY), source=source) + ) + else: + parsed_metadata.pop(PRINCIPAL_METADATA_KEY, None) return ContentLabel( integrity=IntegrityLabel(integrity), confidentiality=ConfidentialityLabel(confidentiality), - metadata=cast(dict[str, Any], metadata) if isinstance(metadata, dict) else None, + metadata=parsed_metadata or None, ) @@ -174,7 +220,11 @@ class ContentLabel(SerializationMixin): user_label = ContentLabel( integrity=IntegrityLabel.TRUSTED, confidentiality=ConfidentialityLabel.USER_IDENTITY, - metadata={"user_id": "user-123"}, + metadata={ + PRINCIPAL_METADATA_KEY: [ + {"tenant_id": "tenant-123", "user_id": "user-123"}, + ] + }, ) """ @@ -244,7 +294,8 @@ def combine_labels(*labels: ContentLabel) -> ContentLabel: The combined label will be: - UNTRUSTED if any input is UNTRUSTED - Most restrictive confidentiality level (USER_IDENTITY > PRIVATE > PUBLIC) - - Merged metadata from all labels + - The union of canonical principal sets for USER_IDENTITY labels + - Merged non-principal metadata from all labels Args: *labels: Variable number of ContentLabel instances to combine. @@ -282,11 +333,33 @@ def combine_labels(*labels: ContentLabel) -> ContentLabel: confidentiality = max((label.confidentiality for label in labels), key=lambda c: confidentiality_priority[c]) - # Merge metadata + # Preserve ordinary metadata behavior while treating identity ownership as a set. merged_metadata: dict[str, Any] = {} for label in labels: if label.metadata: - merged_metadata.update(label.metadata) + merged_metadata.update({ + key: value for key, value in label.metadata.items() if key != PRINCIPAL_METADATA_KEY + }) + + if confidentiality == ConfidentialityLabel.USER_IDENTITY: + principal_sets: list[tuple[tuple[str, str], ...]] = [] + principals_valid = True + for label in labels: + if label.confidentiality != ConfidentialityLabel.USER_IDENTITY: + continue + try: + principal_sets.append( + _canonical_principals( + label.metadata.get(PRINCIPAL_METADATA_KEY), + source="USER_IDENTITY label", + ) + ) + except ValueError: + principals_valid = False + break + if principals_valid and principal_sets: + combined_principals = sorted({principal for principals in principal_sets for principal in principals}) + merged_metadata[PRINCIPAL_METADATA_KEY] = _principal_list(combined_principals) return ContentLabel( integrity=integrity, confidentiality=confidentiality, metadata=merged_metadata if merged_metadata else None @@ -296,6 +369,8 @@ def combine_labels(*labels: ContentLabel) -> ContentLabel: def check_confidentiality_allowed( context_label: ContentLabel, max_allowed: ConfidentialityLabel, + *, + authorized_principals: Sequence[Mapping[str, str]] | None = None, ) -> bool: """Check if writing data with context_label to a destination with max_allowed confidentiality is permitted. @@ -303,12 +378,15 @@ def check_confidentiality_allowed( cannot be written to less secure destinations. For example, it blocks PRIVATE data from being sent to PUBLIC endpoints. - The check passes if context_label.confidentiality <= max_allowed in the hierarchy: + The rank check uses this hierarchy: PUBLIC (0) < PRIVATE (1) < USER_IDENTITY (2) + USER_IDENTITY data additionally requires a valid source principal set that is a subset + of the destination's authorized principals. Args: context_label: The label tracking the confidentiality of data in the current context. max_allowed: The maximum confidentiality level accepted by the destination. + authorized_principals: Principals accepted by a USER_IDENTITY destination. Returns: True if the write is allowed, False if it would be a data exfiltration. @@ -345,7 +423,19 @@ def send_message(destination: str, message: str, context_label: ContentLabel): ConfidentialityLabel.USER_IDENTITY: 2, } - return conf_hierarchy[context_label.confidentiality] <= conf_hierarchy[max_allowed] + if conf_hierarchy[context_label.confidentiality] > conf_hierarchy[max_allowed]: + return False + if context_label.confidentiality != ConfidentialityLabel.USER_IDENTITY: + return True + + try: + source_principals = set( + _canonical_principals(context_label.metadata.get(PRINCIPAL_METADATA_KEY), source="source label") + ) + destination_principals = set(_canonical_principals(authorized_principals, source="destination")) + except ValueError: + return False + return source_principals.issubset(destination_principals) @experimental(feature_id=ExperimentalFeature.FIDES) @@ -1154,8 +1244,8 @@ class LabelTrackingFunctionMiddleware(FunctionMiddleware, _SecurityScopeBinding) +----------+------------------------------------------+----------------------------+ | Priority | Source | When used | +==========+==========================================+============================+ - | Tier 1 | Per-item embedded labels in the result | Always wins if present | - | | (additional_properties.security_label) | | + | Tier 1 | Per-item embedded labels in the result | May only restrict fallback | + | | (additional_properties.security_label) | unless framework-stamped | +----------+------------------------------------------+----------------------------+ | Tier 2 | Tool's source_integrity declaration | No embedded labels | +----------+------------------------------------------+----------------------------+ @@ -1172,10 +1262,11 @@ class LabelTrackingFunctionMiddleware(FunctionMiddleware, _SecurityScopeBinding) 1. Extracts labels from tool input arguments (tier 3 input) 2. Checks tool's source_integrity declaration (tier 2) 3. Executes the tool - 4. Checks for per-item embedded labels in the result (tier 1 — highest priority) + 4. Checks per-item embedded labels against the invocation fallback 5. Falls back to tier 2 or tier 3 when no embedded labels exist - 6. Maintains confidentiality labels based on tool declarations - 7. Automatically hides untrusted content using variable indirection + 6. Accepts complete labels only from identity-stamped framework producers + 7. Maintains confidentiality labels based on tool declarations + 8. Automatically hides untrusted content using variable indirection Attributes: default_integrity: Default integrity for tools without source_integrity declaration. @@ -1311,28 +1402,76 @@ def _extract_primary_tool_content(expanded_content: Any, *, from_quarantined_llm def _resolve_variable_references(self, value: Any) -> tuple[Any, list[ContentLabel]]: """Recursively resolve owned variable references and return their stored labels.""" labels: list[ContentLabel] = [] - return self._resolve_value(value, labels), labels + resolved = self._resolve_value( + value, + labels, + depth=0, + active_variables=set(), + reference_count=[0], + ) + return resolved, labels - def _lookup_variable(self, variable_id: str, labels: list[ContentLabel]) -> Any: + def _lookup_variable( + self, + variable_id: str, + labels: list[ContentLabel], + *, + depth: int, + active_variables: set[str], + reference_count: list[int], + ) -> Any: + if variable_id in active_variables: + raise ValueError("Variable reference cycle detected") + if depth >= _MAX_VARIABLE_REFERENCE_DEPTH: + raise ValueError("Variable reference depth limit exceeded") try: stored_content, stored_label = self.get_variable_store().retrieve(variable_id) except KeyError: return _UNRESOLVED + + reference_count[0] += 1 + if reference_count[0] > _MAX_VARIABLE_REFERENCE_COUNT: + raise ValueError("Variable reference count limit exceeded") labels.append(stored_label) metadata = self.get_variable_metadata(variable_id) - return self._extract_primary_tool_content( + expanded_content = self._extract_primary_tool_content( stored_content, from_quarantined_llm=metadata is not None and metadata.get("function_name") == "quarantined_llm", ) + active_variables.add(variable_id) + try: + return self._resolve_value( + expanded_content, + labels, + depth=depth + 1, + active_variables=active_variables, + reference_count=reference_count, + ) + finally: + active_variables.remove(variable_id) - def _resolve_string(self, value: str, labels: list[ContentLabel]) -> Any: + def _resolve_string( + self, + value: str, + labels: list[ContentLabel], + *, + depth: int, + active_variables: set[str], + reference_count: list[int], + ) -> Any: if not _EMBEDDED_VAR_REF_RE.search(value): return value whole = _WHOLE_VAR_REF_RE.fullmatch(value) if whole is not None: variable_id = whole.group("bracketed") or whole.group("bare") - resolved = self._lookup_variable(variable_id, labels) + resolved = self._lookup_variable( + variable_id, + labels, + depth=depth, + active_variables=active_variables, + reference_count=reference_count, + ) if resolved is _UNRESOLVED: return value if whole.group("bare"): @@ -1341,7 +1480,13 @@ def _resolve_string(self, value: str, labels: list[ContentLabel]) -> Any: def replace(match: re.Match[str]) -> str: variable_id = match.group("bracketed") or match.group("bare") - resolved = self._lookup_variable(variable_id, labels) + resolved = self._lookup_variable( + variable_id, + labels, + depth=depth, + active_variables=active_variables, + reference_count=reference_count, + ) if resolved is _UNRESOLVED: return match.group(0) if match.group("bare"): @@ -1350,18 +1495,59 @@ def replace(match: re.Match[str]) -> str: return _EMBEDDED_VAR_REF_RE.sub(replace, value) - def _resolve_value(self, value: Any, labels: list[ContentLabel]) -> Any: + def _resolve_value( + self, + value: Any, + labels: list[ContentLabel], + *, + depth: int, + active_variables: set[str], + reference_count: list[int], + ) -> Any: if isinstance(value, str): - return self._resolve_string(value, labels) + return self._resolve_string( + value, + labels, + depth=depth, + active_variables=active_variables, + reference_count=reference_count, + ) if isinstance(value, BaseModel): - return self._resolve_value(value.model_dump(), labels) + value = value.model_dump() if isinstance(value, dict): value_dict = cast(dict[str, Any], value) - return {key: self._resolve_value(item, labels) for key, item in value_dict.items()} + return { + key: self._resolve_value( + item, + labels, + depth=depth, + active_variables=active_variables, + reference_count=reference_count, + ) + for key, item in value_dict.items() + } if isinstance(value, list): - return [self._resolve_value(item, labels) for item in cast(list[Any], value)] + return [ + self._resolve_value( + item, + labels, + depth=depth, + active_variables=active_variables, + reference_count=reference_count, + ) + for item in cast(list[Any], value) + ] if isinstance(value, tuple): - return tuple(self._resolve_value(item, labels) for item in cast(tuple[Any, ...], value)) + return tuple( + self._resolve_value( + item, + labels, + depth=depth, + active_variables=active_variables, + reference_count=reference_count, + ) + for item in cast(tuple[Any, ...], value) + ) return value def _expand_variable_references_in_context(self, context: FunctionInvocationContext) -> list[ContentLabel]: @@ -1370,10 +1556,27 @@ def _expand_variable_references_in_context(self, context: FunctionInvocationCont return [] labels: list[ContentLabel] = [] + active_variables: set[str] = set() + reference_count = [0] if context.arguments: - context.arguments = self._resolve_value(context.arguments, labels) + context.arguments = self._resolve_value( + context.arguments, + labels, + depth=0, + active_variables=active_variables, + reference_count=reference_count, + ) if context.kwargs: - context.kwargs = cast(dict[str, Any], self._resolve_value(context.kwargs, labels)) + context.kwargs = cast( + dict[str, Any], + self._resolve_value( + context.kwargs, + labels, + depth=0, + active_variables=active_variables, + reference_count=reference_count, + ), + ) return labels def _get_input_labels(self, context: FunctionInvocationContext) -> list[ContentLabel]: @@ -1552,11 +1755,16 @@ async def process( argument_labels = [*input_labels, *resolved_labels] argument_label = combine_labels(*argument_labels) if argument_labels else ContentLabel() - # Integrity may be declared by the source, but a transformer cannot - # implicitly declassify data derived from its inputs. - result_confidentiality = combine_labels( - ContentLabel(confidentiality=confidentiality), argument_label - ).confidentiality + declared_confidentiality_label = ContentLabel( + confidentiality=confidentiality, + metadata=self._get_function_principal_metadata(context, confidentiality), + ) + result_policy_label = combine_labels(declared_confidentiality_label, argument_label) + result_confidentiality = result_policy_label.confidentiality + + fallback_metadata: dict[str, Any] = {"function_name": function_name} + if PRINCIPAL_METADATA_KEY in result_policy_label.metadata: + fallback_metadata[PRINCIPAL_METADATA_KEY] = result_policy_label.metadata[PRINCIPAL_METADATA_KEY] # Step 3: Build tiered fallback_label # This label is used for result items that have NO embedded labels. @@ -1565,20 +1773,20 @@ async def process( fallback_label = ContentLabel( integrity=declared_source_integrity, confidentiality=result_confidentiality, - metadata={"source": "source_integrity", "function_name": function_name}, + metadata={**fallback_metadata, "source": "source_integrity"}, ) elif argument_labels: combined = combine_labels(*argument_labels) fallback_label = ContentLabel( integrity=combined.integrity, confidentiality=result_confidentiality, - metadata={"source": "input_labels_join", "function_name": function_name}, + metadata={**fallback_metadata, "source": "input_labels_join"}, ) else: fallback_label = ContentLabel( integrity=self.default_integrity, confidentiality=result_confidentiality, - metadata={"source": "default", "function_name": function_name}, + metadata={**fallback_metadata, "source": "default"}, ) context_label = self._context_label @@ -1636,17 +1844,27 @@ def _label_result( # may affect integrity taint. Confidentiality still reflects the most # restrictive label across the entire tool result, including hidden items. if visible_result_label is None: - if result_label.confidentiality != self._context_label.confidentiality: + if ( + result_label.confidentiality != self._context_label.confidentiality + or result_label.confidentiality == ConfidentialityLabel.USER_IDENTITY + ): old_conf = self._context_label.confidentiality hidden_label = ContentLabel( integrity=self._context_label.integrity, confidentiality=result_label.confidentiality, + metadata=result_label.metadata, ) self._update_context_label(hidden_label) logger.info( - f"Result from '{function_name}' hidden (integrity clean) but " - f"confidentiality updated: {old_conf.value} -> " - f"{result_label.confidentiality.value}" + "Result from '%s' hidden (integrity clean) but confidentiality scope updated: %s -> %s", + function_name, + old_conf.value, + result_label.confidentiality.value, + ) + logger.debug( + "Hidden result security metadata merged for '%s': %s", + function_name, + result_label.metadata, ) else: logger.info( @@ -1660,6 +1878,7 @@ def _label_result( exposed_label = ContentLabel( integrity=visible_result_label.integrity, confidentiality=result_label.confidentiality, + metadata=result_label.metadata, ) self._update_context_label(exposed_label) logger.info( @@ -1692,6 +1911,25 @@ def _get_function_confidentiality(self, context: FunctionInvocationContext) -> C return self.default_confidentiality + def _get_function_principal_metadata( + self, + context: FunctionInvocationContext, + confidentiality: ConfidentialityLabel, + ) -> dict[str, Any]: + """Read a locally declared source principal set.""" + if confidentiality != ConfidentialityLabel.USER_IDENTITY: + return {} + function_props = _get_additional_properties(context.function) + try: + principals = _canonical_principals( + function_props.get(PRINCIPAL_METADATA_KEY), + source=f"tool {context.function.name}", + ) + except ValueError as exc: + logger.warning("Invalid USER_IDENTITY source principals for tool '%s': %s", context.function.name, exc) + return {} + return {PRINCIPAL_METADATA_KEY: _principal_list(principals)} + def _process_result_with_embedded_labels( self, items: list[Content], @@ -1700,11 +1938,12 @@ def _process_result_with_embedded_labels( ) -> tuple[list[Content], ContentLabel, ContentLabel | None]: """Process Content items, respecting per-item embedded labels. - This implements the first tier of the label propagation priority: - items with embedded labels (``additional_properties.security_label``) - use those labels directly. Items without embedded labels fall back to - ``fallback_label``, which is either the tool's ``source_integrity`` - declaration (tier 2) or the join of input argument labels (tier 3). + Generic embedded labels (``additional_properties.security_label``) can + only restrict the invocation fallback. A framework-owned producer can + identity-stamp a complete authoritative label after applying local policy. + Items without embedded labels use ``fallback_label``, which is either the + tool's ``source_integrity`` declaration (tier 2) or the join of input + argument labels (tier 3). Each item's own label is attached to its ``additional_properties`` during processing, preserving per-item granularity. @@ -1764,30 +2003,55 @@ def _extract_content_label( The resolved ContentLabel for this item. """ additional_props = _get_additional_properties(item) - authoritative_marker = additional_props.pop(_AUTHORITATIVE_CONFIDENTIALITY, None) + authoritative_marker = additional_props.pop(_AUTHORITATIVE_SECURITY_LABEL, None) + additional_props.pop("_security_label_authoritative_confidentiality", None) inspect_error_marker = additional_props.pop(_INSPECT_VARIABLE_ERROR, None) - authoritative_confidentiality = authoritative_marker is _INTERNAL_RESULT_MARKER + authoritative_label = authoritative_marker is _INTERNAL_RESULT_MARKER inspect_error = function_name == "inspect_variable" and inspect_error_marker is _INTERNAL_RESULT_MARKER label_data = additional_props.get("security_label") if label_data and isinstance(label_data, dict): + label_map = cast(dict[str, Any], label_data) try: embedded_label = _parse_content_label( - cast(dict[str, Any], label_data), + label_map, source="embedded", + allow_principals=authoritative_label, ) + if authoritative_label: + return embedded_label combined_label = combine_labels(fallback_label, embedded_label) - return ContentLabel( - integrity=embedded_label.integrity, - confidentiality=( - embedded_label.confidentiality - if authoritative_confidentiality - else combined_label.confidentiality - ), - metadata=combined_label.metadata, - ) + if ( + combined_label.confidentiality == ConfidentialityLabel.USER_IDENTITY + and fallback_label.confidentiality == ConfidentialityLabel.USER_IDENTITY + ): + try: + fallback_principals = _canonical_principals( + fallback_label.metadata.get(PRINCIPAL_METADATA_KEY), + source="fallback label", + ) + except ValueError: + combined_label.metadata.pop(PRINCIPAL_METADATA_KEY, None) + else: + combined_label.metadata[PRINCIPAL_METADATA_KEY] = _principal_list(fallback_principals) + return combined_label except (TypeError, ValueError) as exc: logger.warning("Failed to parse security_label from Content: %s", exc) + if authoritative_label: + integrity = IntegrityLabel.UNTRUSTED + confidentiality = fallback_label.confidentiality + raw_integrity = label_map.get("integrity") + raw_confidentiality = label_map.get("confidentiality") + if isinstance(raw_integrity, str): + with contextlib.suppress(ValueError): + integrity = IntegrityLabel(raw_integrity) + if isinstance(raw_confidentiality, str): + with contextlib.suppress(ValueError): + confidentiality = ConfidentialityLabel(raw_confidentiality) + return combine_labels( + fallback_label, + ContentLabel(integrity=integrity, confidentiality=confidentiality), + ) if inspect_error: integrity = ( @@ -1955,8 +2219,9 @@ class _PendingPolicyApproval(NamedTuple): approval granted while a placeholder resolved to one payload cannot authorize a replay in which it resolves to something else (or no longer resolves at all); ``label_key`` the conversation label shown for review and ``effective_label_key`` the label of everything the - invocation acts on, including hidden arguments; ``session_key`` the session the approval was - requested in; and ``disclosed_violations`` the canonical risks shown to the user. + invocation acts on, including hidden arguments; ``destination_principal_key`` the locally + declared recipients; ``session_key`` the session the approval was requested in; and + ``disclosed_violations`` the canonical risks shown to the user. ``created_at`` is a wall-clock timestamp so TTL expiration survives session serialization and process restarts. Records remain isolated in the session-scoped security state. """ @@ -1965,6 +2230,7 @@ class _PendingPolicyApproval(NamedTuple): resolved_signature: str label_key: str effective_label_key: str + destination_principal_key: str session_key: str disclosed_violations: tuple[str, ...] request_id: str @@ -1976,19 +2242,21 @@ def to_state(self) -> dict[str, Any]: "resolved_signature": self.resolved_signature, "label_key": self.label_key, "effective_label_key": self.effective_label_key, + "destination_principal_key": self.destination_principal_key, "session_key": self.session_key, "disclosed_violations": list(self.disclosed_violations), "request_id": self.request_id, "created_at": self.created_at, } - def binding_key(self) -> tuple[str, str, str, str, str, tuple[str, ...]]: + def binding_key(self) -> tuple[str, str, str, str, str, str, tuple[str, ...]]: """Return every authorization dimension except lifecycle metadata.""" return ( self.body_signature, self.resolved_signature, self.label_key, self.effective_label_key, + self.destination_principal_key, self.session_key, self.disclosed_violations, ) @@ -2005,6 +2273,7 @@ def from_state(cls, payload: Any) -> _PendingPolicyApproval | None: record["resolved_signature"], record["label_key"], record["effective_label_key"], + record["destination_principal_key"], record["session_key"], ) violations = record["disclosed_violations"] @@ -2023,13 +2292,14 @@ def from_state(cls, payload: Any) -> _PendingPolicyApproval | None: return None if type(created_at) not in (int, float) or not math.isfinite(created_at): return None - typed_values = cast(tuple[str, str, str, str, str], values) + typed_values = cast(tuple[str, str, str, str, str, str], values) return cls( body_signature=typed_values[0], resolved_signature=typed_values[1], label_key=typed_values[2], effective_label_key=typed_values[3], - session_key=typed_values[4], + destination_principal_key=typed_values[4], + session_key=typed_values[5], disclosed_violations=tuple(cast(list[str], violation_items)), request_id=request_id, created_at=float(created_at), @@ -2232,11 +2502,30 @@ def _effective_label_key(self, context: FunctionInvocationContext) -> str: @staticmethod def _label_key(label_data: Any) -> str: if isinstance(label_data, ContentLabel): - return f"{label_data.integrity.value}/{label_data.confidentiality.value}" + principal_key = ( + _principal_binding_key(label_data.metadata.get(PRINCIPAL_METADATA_KEY), source="approval label") + if label_data.confidentiality == ConfidentialityLabel.USER_IDENTITY + else "" + ) + return f"{label_data.integrity.value}/{label_data.confidentiality.value}/{principal_key}" if isinstance(label_data, dict): label_dict = cast(dict[str, Any], label_data) - return f"{label_dict.get('integrity', '')}/{label_dict.get('confidentiality', '')}" - return "/" + confidentiality = label_dict.get("confidentiality", "") + metadata = label_dict.get("metadata") + principal_key = "" + if confidentiality == ConfidentialityLabel.USER_IDENTITY.value: + principal_value = ( + cast(dict[str, Any], metadata).get(PRINCIPAL_METADATA_KEY) if isinstance(metadata, dict) else None + ) + principal_key = _principal_binding_key(principal_value, source="approval label") + return f"{label_dict.get('integrity', '')}/{confidentiality}/{principal_key}" + return "//" + + def _destination_principal_key(self, context: FunctionInvocationContext) -> str: + function_props = _get_additional_properties(context.function) + if function_props.get("max_allowed_confidentiality") != ConfidentialityLabel.USER_IDENTITY.value: + return "" + return _principal_binding_key(function_props.get(PRINCIPAL_METADATA_KEY), source="approval destination") def _session_key(self, context: FunctionInvocationContext) -> str: return context.session.session_id if context.session is not None else "" @@ -2254,6 +2543,7 @@ def _pending_record( resolved_signature=self._resolved_call_signature(context), label_key=self._context_label_key(context), effective_label_key=self._effective_label_key(context), + destination_principal_key=self._destination_principal_key(context), session_key=self._session_key(context), disclosed_violations=self._violation_set_key(violations), request_id=self._get_approval_id(context), @@ -2473,6 +2763,7 @@ async def _process_in_scope( argument_label = self._resolve_label(context.metadata.get("argument_label")) effective_label = combine_labels(context_label, argument_label) + context.metadata["effective_invocation_label"] = effective_label function_props = _get_additional_properties(context.function) accepts_untrusted = ( function_name in self.allow_untrusted_tools or function_props.get("accepts_untrusted") is True @@ -2649,6 +2940,45 @@ def _check_confidentiality_policy_detailed( } except ValueError: logger.warning(f"Invalid max_allowed_confidentiality: {max_allowed_conf}") + if label.confidentiality == ConfidentialityLabel.USER_IDENTITY: + return { + "passed": False, + "failure_type": "principal_mismatch", + "reason": "USER_IDENTITY destination policy is invalid", + } + + if label.confidentiality == ConfidentialityLabel.USER_IDENTITY: + if max_allowed_conf != ConfidentialityLabel.USER_IDENTITY.value: + return { + "passed": False, + "failure_type": "principal_mismatch", + "reason": "USER_IDENTITY destination does not declare an authorized principal set", + } + try: + source_principals = set( + _canonical_principals( + label.metadata.get(PRINCIPAL_METADATA_KEY), + source="source label", + ) + ) + destination_principals = set( + _canonical_principals( + function_props.get(PRINCIPAL_METADATA_KEY), + source=f"tool {context.function.name}", + ) + ) + except ValueError: + return { + "passed": False, + "failure_type": "principal_mismatch", + "reason": "USER_IDENTITY source or destination principals are missing or invalid", + } + if not source_principals.issubset(destination_principals): + return { + "passed": False, + "failure_type": "principal_mismatch", + "reason": "USER_IDENTITY source principals are not authorized for the destination", + } return {"passed": True, "failure_type": None, "reason": None} @@ -3113,24 +3443,31 @@ def _quarantined_llm_result_parser(result: Any) -> list[Content]: logger.warning("quarantined_llm result label is missing confidentiality") return contents + parsed_metadata: dict[str, Any] = {} try: parsed_label = _parse_content_label( typed_label_data, source="quarantined_llm result", + allow_principals=True, ) + parsed_confidentiality = parsed_label.confidentiality + parsed_metadata = parsed_label.metadata except (TypeError, ValueError) as exc: logger.warning("Failed to parse quarantined_llm result label: %s", exc) - return contents + try: + parsed_confidentiality = ConfidentialityLabel(confidentiality) + except ValueError: + parsed_confidentiality = ConfidentialityLabel.PRIVATE quarantine_label = ContentLabel( integrity=IntegrityLabel.UNTRUSTED, - confidentiality=parsed_label.confidentiality, - metadata=parsed_label.metadata, + confidentiality=parsed_confidentiality, + metadata=parsed_metadata, ) first = contents[0] props = first.additional_properties or {} props["security_label"] = quarantine_label.to_dict() - props[_AUTHORITATIVE_CONFIDENTIALITY] = _INTERNAL_RESULT_MARKER + props[_AUTHORITATIVE_SECURITY_LABEL] = _INTERNAL_RESULT_MARKER first.additional_properties = props return contents @@ -3420,6 +3757,7 @@ def _inspect_variable_result_parser(result: Any) -> list[Content]: label = cast(dict[str, Any], result).get("security_label") if isinstance(result, dict) else None if label: props["security_label"] = label + props[_AUTHORITATIVE_SECURITY_LABEL] = _INTERNAL_RESULT_MARKER if isinstance(result, dict) and "error" in cast(dict[str, Any], result): props[_INSPECT_VARIABLE_ERROR] = _INTERNAL_RESULT_MARKER first.additional_properties = props @@ -3866,14 +4204,15 @@ def _stamp_mcp_content_labels( if not isinstance(item, Content): continue props = item.additional_properties or {} - props.pop(_AUTHORITATIVE_CONFIDENTIALITY, None) + props.pop(_AUTHORITATIVE_SECURITY_LABEL, None) + props.pop("_security_label_authoritative_confidentiality", None) server_meta = props.pop(_MCP_RESULT_META_KEY, None) dynamic = _label_from_mcp_meta(server_meta) if server_meta else None if dynamic is None: label = local_label elif trust_server_ifc: label = dynamic - props[_AUTHORITATIVE_CONFIDENTIALITY] = _INTERNAL_RESULT_MARKER + props[_AUTHORITATIVE_SECURITY_LABEL] = _INTERNAL_RESULT_MARKER else: label = combine_labels(local_label, dynamic) props["security_label"] = label.to_dict() diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index d8b500fde98..3c45f2baad6 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -48,6 +48,38 @@ store_untrusted_content, ) +_PRINCIPALS_KEY = "agent_framework.security.principals" +_AUTHORITY_MARKER_KEY = "_security_label_authoritative" + + +def _principal_metadata(user_id: str, tenant_id: str = "tenant-a") -> dict[str, Any]: + return { + _PRINCIPALS_KEY: [ + {"tenant_id": tenant_id, "user_id": user_id}, + ] + } + + +def _identity_destination(principals: Any | None) -> FunctionTool: + class DestinationArgs(BaseModel): + value: str = "value" + + async def destination(value: str = "value") -> str: + return value + + additional_properties: dict[str, Any] = { + "max_allowed_confidentiality": ConfidentialityLabel.USER_IDENTITY.value, + } + if principals is not None: + additional_properties[_PRINCIPALS_KEY] = principals + return FunctionTool( + fn=destination, + name="identity_destination", + description="Identity-scoped destination", + args_schema=DestinationArgs, + additional_properties=additional_properties, + ) + class TestContentLabel: """Tests for ContentLabel class.""" @@ -163,6 +195,23 @@ def test_combine_metadata_merged(self): assert result.metadata["key1"] == "value1" assert result.metadata["key2"] == "value2" + def test_combine_user_identity_principals_uses_set_union(self) -> None: + label_a = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + label_b = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-b"), + ) + + result = combine_labels(label_a, label_b) + + assert result.metadata[_PRINCIPALS_KEY] == [ + {"tenant_id": "tenant-a", "user_id": "user-a"}, + {"tenant_id": "tenant-a", "user_id": "user-b"}, + ] + class TestContentVariableStore: """Tests for ContentVariableStore.""" @@ -298,6 +347,21 @@ def test_store_default_label(self): class TestLabelTrackingMiddleware: """Tests for LabelTrackingFunctionMiddleware.""" + @staticmethod + def _variable_sink() -> FunctionTool: + class SinkArgs(BaseModel): + payload: Any + + async def sink(payload: Any) -> Any: + return payload + + return FunctionTool( + fn=sink, + name="variable_sink", + description="Receive expanded variables", + args_schema=SinkArgs, + ) + @pytest.fixture def middleware(self): """Create middleware instance.""" @@ -360,6 +424,38 @@ async def next_fn(): label = context.metadata["result_label"] assert label.integrity == IntegrityLabel.TRUSTED + async def test_local_source_declaration_preserves_user_identity_principal(self, middleware) -> None: + class SourceArgs(BaseModel): + pass + + async def source() -> str: + return "identity data" + + function = FunctionTool( + fn=source, + name="identity_source", + description="Locally declared identity source", + args_schema=SourceArgs, + additional_properties={ + "source_integrity": "trusted", + "confidentiality": "user_identity", + _PRINCIPALS_KEY: _principal_metadata("user-a")[_PRINCIPALS_KEY], + }, + ) + context = FunctionInvocationContext(function=function, arguments={}) + + async def next_fn() -> None: + context.result = [Content.from_text("identity data")] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].metadata[_PRINCIPALS_KEY] == [ + {"tenant_id": "tenant-a", "user_id": "user-a"} + ] + assert middleware.get_context_label().metadata[_PRINCIPALS_KEY] == [ + {"tenant_id": "tenant-a", "user_id": "user-a"} + ] + @pytest.mark.asyncio async def test_tool_without_source_integrity_defaults_untrusted(self, middleware, mock_function): """Test that tools without source_integrity declaration default to UNTRUSTED.""" @@ -497,6 +593,119 @@ async def next_fn() -> None: await middleware.process(context, next_fn) + async def test_transitive_variable_reference_expands_and_collects_all_labels(self, middleware) -> None: + store = middleware.get_variable_store() + leaf_id = store.store( + "expanded value", + ContentLabel(integrity=IntegrityLabel.UNTRUSTED), + ) + root_id = store.store( + f"[{leaf_id}]", + ContentLabel(integrity=IntegrityLabel.TRUSTED), + ) + function = self._variable_sink() + context = FunctionInvocationContext(function=function, arguments={"payload": f"[{root_id}]"}) + + async def next_fn() -> None: + assert context.arguments == {"payload": "expanded value"} + assert context.metadata["argument_label"].integrity == IntegrityLabel.UNTRUSTED + context.result = [Content.from_text("done")] + + await middleware.process(context, next_fn) + + async def test_three_level_variable_reference_chain_expands(self, middleware) -> None: + store = middleware.get_variable_store() + third_id = store.store("third", ContentLabel(integrity=IntegrityLabel.TRUSTED)) + second_id = store.store(f"[{third_id}]", ContentLabel(integrity=IntegrityLabel.TRUSTED)) + first_id = store.store(f"[{second_id}]", ContentLabel(integrity=IntegrityLabel.TRUSTED)) + function = self._variable_sink() + context = FunctionInvocationContext(function=function, arguments={"payload": f"[{first_id}]"}) + + async def next_fn() -> None: + assert context.arguments == {"payload": "third"} + context.result = [Content.from_text("done")] + + await middleware.process(context, next_fn) + + @pytest.mark.parametrize("indirect", [False, True], ids=["direct", "indirect"]) + async def test_variable_reference_cycle_fails_before_tool_execution(self, middleware, indirect: bool) -> None: + store = middleware.get_variable_store() + first_id = store.store("", ContentLabel(integrity=IntegrityLabel.UNTRUSTED)) + first_entry = cast(dict[str, Any], middleware._scope.variables[first_id]) + if indirect: + second_id = store.store("", ContentLabel(integrity=IntegrityLabel.UNTRUSTED)) + second_entry = cast(dict[str, Any], middleware._scope.variables[second_id]) + first_entry["content"] = f"[{second_id}]" + second_entry["content"] = f"[{first_id}]" + else: + first_entry["content"] = f"[{first_id}]" + + function = self._variable_sink() + context = FunctionInvocationContext(function=function, arguments={"payload": f"[{first_id}]"}) + + async def next_fn() -> None: + pytest.fail("A cyclic variable graph must not execute the destination tool") + + with pytest.raises(ValueError, match="cycle"): + await middleware.process(context, next_fn) + + async def test_nested_mixed_trust_references_expand_and_taint_arguments(self, middleware) -> None: + store = middleware.get_variable_store() + trusted_id = store.store("trusted", ContentLabel(integrity=IntegrityLabel.TRUSTED)) + untrusted_id = store.store("untrusted", ContentLabel(integrity=IntegrityLabel.UNTRUSTED)) + function = self._variable_sink() + context = FunctionInvocationContext( + function=function, + arguments={ + "payload": { + "trusted": f"[{trusted_id}]", + "nested": [f"[{untrusted_id}]", (f"[{trusted_id}]",)], + } + }, + ) + + async def next_fn() -> None: + assert context.arguments == { + "payload": { + "trusted": "trusted", + "nested": ["untrusted", ("trusted",)], + } + } + assert context.metadata["argument_label"].integrity == IntegrityLabel.UNTRUSTED + context.result = [Content.from_text("done")] + + await middleware.process(context, next_fn) + + async def test_variable_reference_depth_limit_fails_before_tool_execution(self, middleware) -> None: + store = middleware.get_variable_store() + value = "leaf" + for _ in range(17): + variable_id = store.store(value, ContentLabel(integrity=IntegrityLabel.TRUSTED)) + value = f"[{variable_id}]" + + function = self._variable_sink() + context = FunctionInvocationContext(function=function, arguments={"payload": value}) + + async def next_fn() -> None: + pytest.fail("An over-depth variable graph must not execute the destination tool") + + with pytest.raises(ValueError, match="depth"): + await middleware.process(context, next_fn) + + async def test_variable_reference_count_limit_fails_before_tool_execution(self, middleware) -> None: + store = middleware.get_variable_store() + references = [ + f"[{store.store(str(index), ContentLabel(integrity=IntegrityLabel.TRUSTED))}]" for index in range(101) + ] + function = self._variable_sink() + context = FunctionInvocationContext(function=function, arguments={"payload": references}) + + async def next_fn() -> None: + pytest.fail("An over-limit variable graph must not execute the destination tool") + + with pytest.raises(ValueError, match="count"): + await middleware.process(context, next_fn) + @pytest.mark.asyncio async def test_json_string_variable_reference_expands_only_response_before_call_next(self, middleware): """JSON-serialized hidden payloads should expose only the response text to tools.""" @@ -1481,7 +1690,9 @@ async def stop_before_execute() -> None: arguments=mock_function.args_schema(arg="test"), ) escalated_context.metadata["context_label"] = ContentLabel( - integrity=IntegrityLabel.UNTRUSTED, confidentiality=ConfidentialityLabel.USER_IDENTITY + integrity=IntegrityLabel.UNTRUSTED, + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), ) escalated_context.metadata["call_id"] = "call-label" escalated_context.metadata["approval_response"] = approval_request.to_function_approval_response(True) @@ -1736,6 +1947,7 @@ async def test_replay_same_violation_type_worse_risk_requires_fresh_approval(sel label = ContentLabel( integrity=IntegrityLabel.TRUSTED, confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), ) request_context = FunctionInvocationContext( function=mock_function, @@ -1821,6 +2033,179 @@ async def execute() -> None: assert isinstance(replay_context.result, Content) assert replay_context.result.type == "function_approval_request" + async def test_approval_for_one_principal_cannot_authorize_another(self) -> None: + middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + function = _identity_destination(_principal_metadata("user-c")[_PRINCIPALS_KEY]) + request_context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + request_context.metadata["context_label"] = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + request_context.metadata["call_id"] = "principal-approval" + + async def stop_before_approval() -> None: + pytest.fail("A mismatched principal flow must require approval") + + with pytest.raises(MiddlewareTermination): + await middleware.process(request_context, stop_before_approval) + + approval_request = request_context.result + assert isinstance(approval_request, Content) + assert approval_request.type == "function_approval_request" + + replay_context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + replay_context.metadata["context_label"] = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-b"), + ) + replay_context.metadata["call_id"] = "principal-approval" + replay_context.metadata["approval_response"] = approval_request.to_function_approval_response(True) + executed = False + + async def execute() -> None: + nonlocal executed + executed = True + + with pytest.raises(MiddlewareTermination): + await middleware.process(replay_context, execute) + + assert executed is False + assert isinstance(replay_context.result, Content) + assert replay_context.result.type == "function_approval_request" + assert replay_context.result.additional_properties["_replacement_approval_request"] is True + + async def test_approval_is_bound_to_destination_principals(self) -> None: + middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + function = _identity_destination(_principal_metadata("user-b")[_PRINCIPALS_KEY]) + source_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + request_context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + request_context.metadata["context_label"] = source_label + request_context.metadata["call_id"] = "destination-principal-approval" + + async def stop_before_approval() -> None: + pytest.fail("A mismatched principal flow must require approval") + + with pytest.raises(MiddlewareTermination): + await middleware.process(request_context, stop_before_approval) + + approval_request = request_context.result + assert isinstance(approval_request, Content) + assert function.additional_properties is not None + function.additional_properties[_PRINCIPALS_KEY] = _principal_metadata("user-c")[_PRINCIPALS_KEY] + + replay_context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + replay_context.metadata["context_label"] = source_label + replay_context.metadata["call_id"] = "destination-principal-approval" + replay_context.metadata["approval_response"] = approval_request.to_function_approval_response(True) + executed = False + + async def execute() -> None: + nonlocal executed + executed = True + + with pytest.raises(MiddlewareTermination): + await middleware.process(replay_context, execute) + + assert executed is False + assert isinstance(replay_context.result, Content) + assert replay_context.result.type == "function_approval_request" + assert replay_context.result.additional_properties["_replacement_approval_request"] is True + + @pytest.mark.parametrize("malformed_source", [True, False], ids=["source", "destination"]) + async def test_malformed_principals_cannot_create_approval_authority(self, malformed_source: bool) -> None: + middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + destination_principals: Any = ( + _principal_metadata("user-b")[_PRINCIPALS_KEY] + if malformed_source + else {"tenant_id": "tenant-a", "user_id": "user-b"} + ) + source_metadata = {} if malformed_source else _principal_metadata("user-a") + function = _identity_destination(destination_principals) + context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + context.metadata["context_label"] = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=source_metadata, + ) + context.metadata["call_id"] = "invalid-principal-approval" + + async def execute() -> None: + pytest.fail("Malformed principal state must not create approval authority") + + with pytest.raises(MiddlewareTermination): + await middleware.process(context, execute) + + assert isinstance(context.result, dict) + assert context.result["violation_type"] == "unsafe_approval_binding" + + async def test_approval_binds_computed_argument_principals_without_label_tracker(self) -> None: + middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + function = _identity_destination(_principal_metadata("user-c")[_PRINCIPALS_KEY]) + request_context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + request_context.metadata.update({ + "context_label": ContentLabel(), + "argument_label": ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ), + "call_id": "argument-principal-approval", + }) + + async def stop_before_approval() -> None: + pytest.fail("A mismatched principal flow must require approval") + + with pytest.raises(MiddlewareTermination): + await middleware.process(request_context, stop_before_approval) + + approval_request = request_context.result + assert isinstance(approval_request, Content) + + replay_context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + replay_context.metadata.update({ + "context_label": ContentLabel(), + "argument_label": ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-b"), + ), + "call_id": "argument-principal-approval", + "approval_response": approval_request.to_function_approval_response(True), + }) + executed = False + + async def execute() -> None: + nonlocal executed + executed = True + + with pytest.raises(MiddlewareTermination): + await middleware.process(replay_context, execute) + + assert executed is False + assert isinstance(replay_context.result, Content) + assert replay_context.result.additional_properties["_replacement_approval_request"] is True + class TestAutomaticHiding: """Tests for automatic variable hiding functionality.""" @@ -2105,7 +2490,7 @@ async def test_inspect_variable_propagates_user_identity(self, middleware_no_aut ContentLabel( integrity=IntegrityLabel.UNTRUSTED, confidentiality=ConfidentialityLabel.USER_IDENTITY, - metadata={"user_id": "user-123"}, + metadata=_principal_metadata("user-123"), ), ) @@ -3318,6 +3703,7 @@ async def test_quarantined_llm_publishes_combined_confidentiality(self) -> None: ContentLabel( integrity=IntegrityLabel.TRUSTED, confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), ), ) quarantine_tool = next(tool for tool in middleware.get_security_tools() if tool.name == "quarantined_llm") @@ -3340,6 +3726,31 @@ async def next_fn() -> None: assert hidden_label.integrity == IntegrityLabel.UNTRUSTED assert hidden_label.confidentiality == ConfidentialityLabel.USER_IDENTITY + async def test_quarantined_llm_preserves_user_identity_when_principals_are_missing(self) -> None: + middleware = LabelTrackingFunctionMiddleware() + variable_id = middleware.get_variable_store().store( + "identity secret", + ContentLabel( + integrity=IntegrityLabel.UNTRUSTED, + confidentiality=ConfidentialityLabel.USER_IDENTITY, + ), + ) + quarantine_tool = next(tool for tool in middleware.get_security_tools() if tool.name == "quarantined_llm") + context = FunctionInvocationContext( + function=quarantine_tool, + arguments={"prompt": "Summarize", "variable_ids": [variable_id]}, + ) + + async def next_fn() -> None: + context.result = await quarantine_tool.invoke(arguments=context.arguments, context=context) + + await middleware.process(context, next_fn) + + result_label = context.metadata["result_label"] + assert result_label.integrity == IntegrityLabel.UNTRUSTED + assert result_label.confidentiality == ConfidentialityLabel.USER_IDENTITY + assert _PRINCIPALS_KEY not in result_label.metadata + async def test_quarantined_llm_public_input_remains_public(self) -> None: """A valid quarantine label overrides the fail-closed PRIVATE fallback.""" from agent_framework.security import set_quarantine_client @@ -3886,9 +4297,167 @@ async def mock_fn() -> list: return FunctionTool(fn=mock_fn, name="fetch_items", description="Fetch items", args_schema=MockArgs) + async def test_untrusted_fallback_cannot_be_upgraded_by_embedded_label(self, middleware, mock_function) -> None: + context = FunctionInvocationContext(function=mock_function, arguments=mock_function.args_schema()) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "remote result", + additional_properties={ + "security_label": {"integrity": "trusted", "confidentiality": "public"}, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].integrity == IntegrityLabel.UNTRUSTED + assert context.result[0].additional_properties["_variable_reference"] is True + + async def test_trusted_fallback_can_be_restricted_by_embedded_label(self, middleware, mock_function) -> None: + mock_function.additional_properties = {"source_integrity": "trusted"} + context = FunctionInvocationContext(function=mock_function, arguments=mock_function.args_schema()) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "remote result", + additional_properties={ + "security_label": {"integrity": "untrusted", "confidentiality": "public"}, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].integrity == IntegrityLabel.UNTRUSTED + assert context.result[0].additional_properties["_variable_reference"] is True + + async def test_framework_authoritative_trusted_label_is_preserved(self, middleware, mock_function) -> None: + from agent_framework.security import _INTERNAL_RESULT_MARKER + + context = FunctionInvocationContext(function=mock_function, arguments=mock_function.args_schema()) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "locally verified result", + additional_properties={ + "security_label": {"integrity": "trusted", "confidentiality": "public"}, + _AUTHORITY_MARKER_KEY: _INTERNAL_RESULT_MARKER, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].integrity == IntegrityLabel.TRUSTED + assert context.result[0].text == "locally verified result" + assert _AUTHORITY_MARKER_KEY not in context.result[0].additional_properties + + @pytest.mark.parametrize("forged_marker", [True, "framework"], ids=["boolean", "string"]) + async def test_serializable_authority_marker_is_ignored( + self, + middleware, + mock_function, + forged_marker: bool | str, + ) -> None: + context = FunctionInvocationContext(function=mock_function, arguments=mock_function.args_schema()) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "remote result", + additional_properties={ + "security_label": {"integrity": "trusted", "confidentiality": "public"}, + _AUTHORITY_MARKER_KEY: forged_marker, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].integrity == IntegrityLabel.UNTRUSTED + assert context.result[0].additional_properties["_variable_reference"] is True + + async def test_embedded_label_cannot_reduce_fallback_confidentiality(self, middleware, mock_function) -> None: + mock_function.additional_properties = {"source_integrity": "trusted", "confidentiality": "private"} + context = FunctionInvocationContext(function=mock_function, arguments=mock_function.args_schema()) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "remote result", + additional_properties={ + "security_label": {"integrity": "trusted", "confidentiality": "public"}, + _AUTHORITY_MARKER_KEY: True, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].confidentiality == ConfidentialityLabel.PRIVATE + assert _AUTHORITY_MARKER_KEY not in context.result[0].additional_properties + + async def test_generic_embedded_label_cannot_introduce_principals(self, middleware, mock_function) -> None: + mock_function.additional_properties = {"source_integrity": "trusted"} + context = FunctionInvocationContext(function=mock_function, arguments=mock_function.args_schema()) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "remote identity data", + additional_properties={ + "security_label": { + "integrity": "trusted", + "confidentiality": "user_identity", + "metadata": _principal_metadata("user-a"), + }, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].confidentiality == ConfidentialityLabel.USER_IDENTITY + assert _PRINCIPALS_KEY not in context.metadata["result_label"].metadata + + async def test_generic_identity_label_preserves_authoritative_fallback_principals( + self, middleware, mock_function + ) -> None: + mock_function.additional_properties = { + "source_integrity": "trusted", + "confidentiality": "user_identity", + _PRINCIPALS_KEY: _principal_metadata("user-a")[_PRINCIPALS_KEY], + } + context = FunctionInvocationContext(function=mock_function, arguments={}) + + async def next_fn() -> None: + context.result = [ + Content.from_text( + "remote identity data", + additional_properties={ + "security_label": { + "integrity": "trusted", + "confidentiality": "user_identity", + "metadata": _principal_metadata("user-b"), + }, + }, + ) + ] + + await middleware.process(context, next_fn) + + assert context.metadata["result_label"].metadata[_PRINCIPALS_KEY] == [ + {"tenant_id": "tenant-a", "user_id": "user-a"} + ] + @pytest.mark.asyncio async def test_mixed_trust_items_in_list(self, middleware, mock_function): """Test that untrusted items are hidden while trusted items remain visible.""" + mock_function.additional_properties = {"source_integrity": "trusted"} args = mock_function.args_schema() context = FunctionInvocationContext(function=mock_function, arguments=args) @@ -3941,6 +4510,7 @@ async def next_fn(): @pytest.mark.asyncio async def test_hidden_untrusted_items_do_not_taint_integrity_in_mixed_results(self, middleware, mock_function): """Hidden untrusted items should only affect confidentiality, not integrity.""" + mock_function.additional_properties = {"source_integrity": "trusted"} args = mock_function.args_schema() context = FunctionInvocationContext(function=mock_function, arguments=args) @@ -3965,6 +4535,7 @@ async def next_fn(): @pytest.mark.asyncio async def test_all_trusted_items_visible(self, middleware, mock_function): """Test that all trusted items remain fully visible.""" + mock_function.additional_properties = {"source_integrity": "trusted"} args = mock_function.args_schema() context = FunctionInvocationContext(function=mock_function, arguments=args) @@ -4044,6 +4615,40 @@ async def next_fn() -> None: assert middleware.get_context_label().integrity == IntegrityLabel.TRUSTED assert middleware.get_context_label().confidentiality == ConfidentialityLabel.USER_IDENTITY + async def test_hidden_user_identity_result_unions_principals_at_same_rank(self, middleware, mock_function) -> None: + middleware._context_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + mock_function.additional_properties = { + "source_integrity": "untrusted", + "confidentiality": "user_identity", + _PRINCIPALS_KEY: _principal_metadata("user-b")[_PRINCIPALS_KEY], + } + context = FunctionInvocationContext(function=mock_function, arguments={}) + + async def next_fn() -> None: + context.result = [Content.from_text("hidden user B data")] + + await middleware.process(context, next_fn) + + assert context.result[0].additional_properties["_variable_reference"] is True + assert middleware.get_context_label().metadata[_PRINCIPALS_KEY] == [ + {"tenant_id": "tenant-a", "user_id": "user-a"}, + {"tenant_id": "tenant-a", "user_id": "user-b"}, + ] + + destination = _identity_destination(_principal_metadata("user-a")[_PRINCIPALS_KEY]) + policy_context = FunctionInvocationContext(function=destination, arguments={}) + policy_context.metadata["context_label"] = middleware.get_context_label() + policy = PolicyEnforcementFunctionMiddleware() + + async def execute() -> None: + pytest.fail("A single-user destination must not receive hidden content from another principal") + + with pytest.raises(MiddlewareTermination): + await policy.process(policy_context, execute) + async def test_malformed_embedded_metadata_preserves_mandatory_label_fields( self, middleware, mock_function ) -> None: @@ -4390,16 +4995,17 @@ async def next_fn() -> None: "transformed", additional_properties={ "security_label": {"integrity": "trusted", "confidentiality": "public"}, - "_security_label_authoritative_confidentiality": True, + _AUTHORITY_MARKER_KEY: True, }, ) ] await middleware.process(context, next_fn) - assert context.metadata["result_label"].integrity == IntegrityLabel.TRUSTED + assert context.metadata["result_label"].integrity == IntegrityLabel.UNTRUSTED assert context.metadata["result_label"].confidentiality == ConfidentialityLabel.PRIVATE assert middleware.get_context_label().confidentiality == ConfidentialityLabel.PRIVATE + assert _AUTHORITY_MARKER_KEY not in context.result[0].additional_properties @pytest.mark.asyncio async def test_embedded_labels_override_source_integrity(self, middleware): @@ -4695,6 +5301,96 @@ async def next_fn(): # Should be blocked (either violation should block) assert "error" in context.result + async def test_matching_user_identity_principal_is_allowed(self, policy_middleware) -> None: + source_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + function = _identity_destination(_principal_metadata("user-a")[_PRINCIPALS_KEY]) + context = FunctionInvocationContext(function=function, arguments={}) + context.metadata["context_label"] = source_label + + async def next_fn() -> None: + context.result = "sent" + + await policy_middleware.process(context, next_fn) + + assert context.result == "sent" + + @pytest.mark.parametrize( + ("source_metadata", "destination_principals"), + [ + (_principal_metadata("user-a"), _principal_metadata("user-b")[_PRINCIPALS_KEY]), + ( + _principal_metadata("user-a", "tenant-a"), + _principal_metadata("user-a", "tenant-b")[_PRINCIPALS_KEY], + ), + ({}, _principal_metadata("user-a")[_PRINCIPALS_KEY]), + (_principal_metadata("user-a"), None), + ( + {_PRINCIPALS_KEY: {"tenant_id": "tenant-a", "user_id": "user-a"}}, + _principal_metadata("user-a")[_PRINCIPALS_KEY], + ), + ( + _principal_metadata("user-a"), + {"tenant_id": "tenant-a", "user_id": "user-a"}, + ), + ], + ids=[ + "different-user", + "different-tenant", + "missing-source", + "missing-destination", + "malformed-source", + "malformed-destination", + ], + ) + async def test_user_identity_principal_mismatch_is_blocked( + self, + policy_middleware, + source_metadata: dict[str, Any], + destination_principals: Any | None, + ) -> None: + source_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=source_metadata, + ) + function = _identity_destination(destination_principals) + context = FunctionInvocationContext(function=function, arguments={}) + context.metadata["context_label"] = source_label + executed = False + + async def next_fn() -> None: + nonlocal executed + executed = True + + with pytest.raises(MiddlewareTermination): + await policy_middleware.process(context, next_fn) + + assert executed is False + assert context.result["violation_type"] == "principal_mismatch" + + async def test_combined_principals_require_destination_subset(self, policy_middleware) -> None: + source_label = combine_labels( + ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ), + ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-b"), + ), + ) + function = _identity_destination(_principal_metadata("user-a")[_PRINCIPALS_KEY]) + context = FunctionInvocationContext(function=function, arguments={}) + context.metadata["context_label"] = source_label + + async def next_fn() -> None: + pytest.fail("A destination for one owner must not receive content owned by multiple principals") + + with pytest.raises(MiddlewareTermination): + await policy_middleware.process(context, next_fn) + class TestCheckConfidentialityAllowed: """Tests for check_confidentiality_allowed helper function.""" @@ -4755,12 +5451,55 @@ def test_user_identity_to_private_blocked(self): ui_label = ContentLabel(confidentiality=ConfidentialityLabel.USER_IDENTITY) assert check_confidentiality_allowed(ui_label, ConfidentialityLabel.PRIVATE) is False - def test_user_identity_to_user_identity_allowed(self): - """Test USER_IDENTITY data can be written to USER_IDENTITY destination.""" + def test_matching_user_identity_to_user_identity_allowed(self): from agent_framework.security import check_confidentiality_allowed - ui_label = ContentLabel(confidentiality=ConfidentialityLabel.USER_IDENTITY) - assert check_confidentiality_allowed(ui_label, ConfidentialityLabel.USER_IDENTITY) is True + ui_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + assert ( + check_confidentiality_allowed( + ui_label, + ConfidentialityLabel.USER_IDENTITY, + authorized_principals=_principal_metadata("user-a")[_PRINCIPALS_KEY], + ) + is True + ) + + @pytest.mark.parametrize( + ("source_metadata", "authorized_principals"), + [ + (_principal_metadata("user-a"), _principal_metadata("user-b")[_PRINCIPALS_KEY]), + ( + _principal_metadata("user-a", "tenant-a"), + _principal_metadata("user-a", "tenant-b")[_PRINCIPALS_KEY], + ), + ({}, _principal_metadata("user-a")[_PRINCIPALS_KEY]), + (_principal_metadata("user-a"), None), + ], + ids=["different-user", "different-tenant", "missing-source", "missing-destination"], + ) + def test_user_identity_requires_matching_principals( + self, + source_metadata: dict[str, Any], + authorized_principals: Any | None, + ) -> None: + from agent_framework.security import check_confidentiality_allowed + + ui_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=source_metadata, + ) + + assert ( + check_confidentiality_allowed( + ui_label, + ConfidentialityLabel.USER_IDENTITY, + authorized_principals=authorized_principals, + ) + is False + ) if __name__ == "__main__": @@ -5363,6 +6102,31 @@ async def test_apply_mcp_security_labels_configures_result_authority(self, trust ) assert result[0].additional_properties["security_label"] == expected_label + async def test_framework_stamped_mcp_label_remains_authoritative_through_tracking(self) -> None: + from agent_framework.security import apply_mcp_security_labels + + annotations = SimpleNamespace(readOnlyHint=True, openWorldHint=False) + server_meta = {"ifc": {"integrity": "trusted", "confidentiality": "public"}} + mcp_tool, function = _make_connected_mcp_tool_for_ifc( + annotations=annotations, + server_meta=server_meta, + ) + await apply_mcp_security_labels(mcp_tool, trust_server_ifc=True) + assert function.func is not None + stamped_result = await function.func() + tracker = LabelTrackingFunctionMiddleware() + context = FunctionInvocationContext(function=function, arguments={}) + + async def next_fn() -> None: + context.result = stamped_result + + await tracker.process(context, next_fn) + + assert context.metadata["result_label"].integrity == IntegrityLabel.TRUSTED + assert context.result[0].text == "payload" + assert _AUTHORITY_MARKER_KEY not in context.result[0].additional_properties + assert "_security_label_authoritative_confidentiality" not in context.result[0].additional_properties + async def test_apply_mcp_security_labels_reconfigures_existing_wrapper_authority(self): from agent_framework.security import apply_mcp_security_labels @@ -5588,6 +6352,58 @@ async def execute(current: FunctionInvocationContext) -> list[Content]: ) == ["alice secret", "bob secret"] assert get_current_middleware() is None + async def test_overlapping_sessions_keep_principal_metadata_isolated(self) -> None: + tracker = LabelTrackingFunctionMiddleware(auto_hide_untrusted=False) + alice = AgentSession(session_id="alice-principal") + bob = AgentSession(session_id="bob-principal") + tracker._scope_for_session(alice).context_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-a"), + ) + tracker._scope_for_session(bob).context_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata=_principal_metadata("user-b"), + ) + + class SourceArgs(BaseModel): + pass + + async def source() -> str: + return "done" + + function = FunctionTool( + fn=source, + name="session_principal_source", + description="Observe session principal", + args_schema=SourceArgs, + additional_properties={"source_integrity": "trusted"}, + ) + both_started = asyncio.Event() + started = 0 + + async def run(session: AgentSession, expected_user_id: str) -> None: + nonlocal started + context = FunctionInvocationContext(function=function, arguments={}, session=session) + + async def execute() -> None: + nonlocal started + started += 1 + if started == 2: + both_started.set() + await both_started.wait() + await asyncio.sleep(0) + assert context.metadata["context_label"].metadata[_PRINCIPALS_KEY] == [ + {"tenant_id": "tenant-a", "user_id": expected_user_id} + ] + context.result = [Content.from_text("done")] + + await tracker.process(context, execute) + + await asyncio.gather( + run(alice, "user-a"), + run(bob, "user-b"), + ) + async def test_direct_standalone_invocation_keeps_private_scope(self) -> None: tracker = LabelTrackingFunctionMiddleware() policy = PolicyEnforcementFunctionMiddleware() diff --git a/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py index b8491f2ea79..fc1f9fc3776 100644 --- a/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py +++ b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py @@ -15,6 +15,7 @@ WorkflowRunResult, executor, ) +from agent_framework.security import ContentLabel, IntegrityLabel, LabelTrackingFunctionMiddleware from google.protobuf.json_format import MessageToDict, ParseDict from pytest import raises @@ -77,6 +78,38 @@ def test_a2a_to_run_converts_supported_parts() -> None: assert converted.contents[3].text == '"structured"' +def test_a2a_security_label_metadata_cannot_upgrade_local_integrity() -> None: + message = A2AMessage( + message_id="message-security-label", + role=Role.ROLE_USER, + parts=[ + Part( + text="remote content", + metadata={ + "security_label": { + "integrity": "trusted", + "confidentiality": "public", + } + }, + ) + ], + ) + + run = a2a_to_run(message) + messages = run["messages"] + assert isinstance(messages, list) + converted_message = messages[0] + assert isinstance(converted_message, Message) + converted = converted_message.contents[0] + middleware = LabelTrackingFunctionMiddleware() + processed, label, _ = middleware._process_result_with_embedded_labels( + [converted], "a2a_remote", ContentLabel(integrity=IntegrityLabel.UNTRUSTED) + ) + + assert label.integrity == IntegrityLabel.UNTRUSTED + assert processed[0].additional_properties["_variable_reference"] is True + + def test_a2a_to_run_rejects_empty_message() -> None: with raises(ValueError, match="no supported"): a2a_to_run(A2AMessage(message_id="message-1", role=Role.ROLE_USER)) From 6aabf24d667e6a6031b0fbdcf2ff5ae130184d58 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 10 Sep 2026 14:57:46 +0200 Subject: [PATCH 2/3] Python: add identity security setup sample Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../security/FIDES_DEVELOPER_GUIDE.md | 97 ++++++++-- python/samples/02-agents/security/README.md | 20 ++ .../security/email_security_example.py | 10 +- .../user_identity_security_example.py | 172 ++++++++++++++++++ 4 files changed, 284 insertions(+), 15 deletions(-) create mode 100644 python/samples/02-agents/security/user_identity_security_example.py diff --git a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md index 9ebf7bb19c9..8f7efb0dd3d 100644 --- a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md +++ b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md @@ -43,33 +43,86 @@ Every piece of content (tool calls, results, messages) can be assigned a `Conten - **USER_IDENTITY**: Content is restricted to specific user identities only ```python -from agent_framework.security import ContentLabel, IntegrityLabel, ConfidentialityLabel +from agent_framework.security import ConfidentialityLabel, ContentLabel, IntegrityLabel, PRINCIPAL_METADATA_KEY # Create a label label = ContentLabel( integrity=IntegrityLabel.TRUSTED, - confidentiality=ConfidentialityLabel.PRIVATE, - metadata={"user_id": "user-123"} + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata={ + PRINCIPAL_METADATA_KEY: [ + {"tenant_id": "tenant-123", "user_id": "user-123"}, + ] + }, ) ``` +### 1.1 USER_IDENTITY Principal Binding + +`USER_IDENTITY` labels require a canonical, non-empty principal set. Each +principal contains exactly `tenant_id` and `user_id`. Build this metadata from +the authenticated request or session, or from a locally trusted static tool +declaration. Do not infer it from model arguments or remote result metadata. + +Source tools declare the owner of identity-scoped output. Destination tools +declare the principals they authorize using the same namespaced key: + +```python +from agent_framework import tool +from agent_framework.security import PRINCIPAL_METADATA_KEY + +alice = [{"tenant_id": "tenant-contoso", "user_id": "alice"}] + + +@tool( + description="Read Alice's profile", + additional_properties={ + "source_integrity": "trusted", + "confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: alice, + }, +) +async def read_profile() -> str: + return "Alice profile data" + + +@tool( + description="Save data to Alice's profile", + additional_properties={ + "max_allowed_confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: alice, + }, +) +async def save_profile(data: str) -> None: + ... +``` + +The policy allows a flow only when every source principal is present in the +destination's authorized set. Missing, malformed, or mismatched principal data +is blocked. See +[`user_identity_security_example.py`](user_identity_security_example.py) for a +complete runnable setup. + ### 2. Label Tracking Middleware with Tiered Label Propagation `LabelTrackingFunctionMiddleware` uses a **tiered label propagation** scheme where the result label of a tool call is determined by a strict 3-tier priority: | Priority | Source | Used When | |----------|--------|-----------| -| **Tier 1** (Highest) | Per-item embedded labels (`additional_properties.security_label`) | Tool result items include explicit labels | +| **Tier 1** | Per-item embedded labels (`additional_properties.security_label`) | Restrict the locally established fallback | | **Tier 2** | Tool's `source_integrity` declaration | No embedded labels, but tool declares `source_integrity` | | **Tier 3** (Lowest) | Join of input argument labels (`combine_labels`) | No embedded labels AND no `source_integrity` declared | | **Default** | `UNTRUSTED` | No labels from any tier | **Tiered Label Propagation:** -- **Tier 1: Embedded labels** in result items via `additional_properties.security_label` — highest priority, used per-item -- **Tier 2: `source_integrity`** declaration on the tool — authoritative for the trust level of the tool's output, regardless of input labels +- **Tier 1: Embedded labels** are restriction-only by default: they can downgrade integrity or raise confidentiality, but cannot upgrade a fallback or supply principal authority +- **Tier 2: `source_integrity`** is the locally trusted fallback for the tool's output; use `"trusted"` only after the local connector enforces its trust policy - **Tier 3: Input labels join** — `combine_labels(*input_labels)` from arguments (VariableReferenceContent, labeled data) - **Default**: `UNTRUSTED` when no labels exist from any tier +Framework-owned parsers and wrappers may stamp a complete label after enforcing +local policy. Application and remote metadata remains restriction-only. + **Per-Item Embedded Labels (RECOMMENDED for Mixed-Trust Data):** Tools returning mixed-trust data should embed labels on each item in `additional_properties.security_label`: @@ -88,7 +141,7 @@ The middleware automatically: **Tool-Level Source Integrity (Tier 2 Fallback):** If items don't have embedded labels, the tool can declare a fallback via `source_integrity`. -When declared, `source_integrity` alone determines the result label — input argument labels are NOT combined in. This means a tool declaring `source_integrity="trusted"` always produces trusted output regardless of what inputs it received: +When declared, `source_integrity` establishes the local integrity fallback. Embedded labels can make the result less trusted, but cannot make it more trusted: - `source_integrity="trusted"`: Tool produces trusted data (internal computations) - `source_integrity="untrusted"`: Tool fetches untrusted data - (not set): Falls back to tier 3 (join of input labels) or **UNTRUSTED** default @@ -112,7 +165,12 @@ from agent_framework import Content, tool from agent_framework.security import LabelTrackingFunctionMiddleware, SecureAgentConfig # Define a tool that returns mixed-trust data with per-item labels -@tool(description="Fetch emails from inbox") +@tool( + description="Fetch emails from inbox", + additional_properties={ + "source_integrity": "trusted", # Local connector verifies internal senders + }, +) async def fetch_emails(count: int = 5) -> list[Content]: """Fetch emails - some from trusted internal sources, others from external sources.""" emails = get_emails(count) @@ -171,7 +229,12 @@ For tools that return mixed-trust data (e.g., emails from both internal and exte import json from agent_framework import Content, tool -@tool(description="Fetch emails from inbox") +@tool( + description="Fetch emails from inbox", + additional_properties={ + "source_integrity": "trusted", # Local connector verifies internal senders + }, +) async def fetch_emails(count: int = 5) -> list[Content]: """Fetch emails with per-item security labels.""" emails = fetch_from_server(count) @@ -685,7 +748,12 @@ import json from agent_framework import Content, tool # Tool returning mixed-trust data with per-item labels (RECOMMENDED) -@tool(description="Fetch emails from inbox") +@tool( + description="Fetch emails from inbox", + additional_properties={ + "source_integrity": "trusted", # Local connector verifies internal senders + }, +) async def fetch_emails(count: int = 5) -> list[Content]: """Emails can be from trusted internal or untrusted external sources.""" emails = get_emails(count) @@ -836,7 +904,7 @@ PUBLIC (0) < PRIVATE (1) < USER_IDENTITY (2) - PUBLIC data can flow anywhere - PRIVATE data can only flow to PRIVATE or USER_IDENTITY destinations -- USER_IDENTITY data can only flow to USER_IDENTITY destinations +- USER_IDENTITY data can only flow to USER_IDENTITY destinations that authorize every source principal **Runtime Helper Function:** @@ -893,8 +961,10 @@ await post_to_slack(channel="#docs", message="Check out our docs!") |----------|---------|----------------| | `confidentiality` | Declares output sensitivity | `"public"`, `"private"`, `"user_identity"` | | `max_allowed_confidentiality` | Gates outputs (maximum level) | `"public"` = blocks PRIVATE data exfiltration | +| `agent_framework.security.principals` | Declares USER_IDENTITY owners or authorized destinations | `[{"tenant_id": "tenant-contoso", "user_id": "alice"}]` | -See `samples/02-agents/security/repo_confidentiality_example.py` for a complete working example. +See `repo_confidentiality_example.py` for confidentiality ranking and +`user_identity_security_example.py` for principal-bound identity data. ## Configuration Options @@ -1021,6 +1091,7 @@ Run the maintained security samples from `python/`: ```bash uv run samples/02-agents/security/email_security_example.py --cli uv run samples/02-agents/security/repo_confidentiality_example.py --cli +uv run samples/02-agents/security/user_identity_security_example.py uv run samples/02-agents/security/github_mcp_example.py --cli uv run samples/02-agents/security/github_mcp_example.py --cli --attack ``` @@ -1031,6 +1102,7 @@ This demonstrates: - Quarantined LLM usage - Variable inspection - Policy enforcement +- Principal-bound USER_IDENTITY sources and destinations - Complete secure workflow ## Key Takeaways @@ -1059,6 +1131,7 @@ from agent_framework.security import ( ContentLabel, IntegrityLabel, ConfidentialityLabel, + PRINCIPAL_METADATA_KEY, combine_labels, # Variable Store diff --git a/python/samples/02-agents/security/README.md b/python/samples/02-agents/security/README.md index 54cd0e46bf5..d3105eb38ff 100644 --- a/python/samples/02-agents/security/README.md +++ b/python/samples/02-agents/security/README.md @@ -11,6 +11,7 @@ security model, middleware behavior, and API reference. |--------|-------|--------------| | `email_security_example.py` | Prompt injection defense | `SecureAgentConfig`, Foundry-backed email handling, `quarantined_llm`, and approval on policy violations | | `repo_confidentiality_example.py` | Data exfiltration prevention | Confidentiality labels, Foundry-backed repository access, `max_allowed_confidentiality`, and approval before leaking private data | +| `user_identity_security_example.py` | Principal-bound identity data | Canonical tenant/user principals, identity-scoped sources and destinations, and allowed versus blocked flows | | `github_mcp_example.py` | Remote MCP URL with local FIDES enforcement | `SecureMCPToolProxy(url=...)`, direct GitHub MCP access, tool auto-labeling, and post-tool-call policy enforcement | ## Prerequisites @@ -87,6 +88,25 @@ What to look for: - Reading private content taints the context as private - Posting private data to a public destination triggers an approval request +### `user_identity_security_example.py` + +This sample creates source and destination tools from a host-authenticated +tenant/user principal. It demonstrates the canonical principal metadata format +and wires policy enforcement with `SecureAgentConfig`. + +Run it with: + +```bash +uv run samples/02-agents/security/user_identity_security_example.py +``` + +What to look for: + +- Data read for Alice carries Alice's tenant/user principal +- Saving that data to Alice's destination is allowed +- Sending the same data to Bob's destination is blocked and audited +- Principal identity comes from host configuration, not model tool arguments + ### `github_mcp_example.py` This sample connects directly to `https://api.githubcopilot.com/mcp/` through diff --git a/python/samples/02-agents/security/email_security_example.py b/python/samples/02-agents/security/email_security_example.py index f8dff526ff4..14f7aacd02a 100644 --- a/python/samples/02-agents/security/email_security_example.py +++ b/python/samples/02-agents/security/email_security_example.py @@ -204,7 +204,11 @@ async def send_email( @tool( description="Fetch emails from the inbox. Returns a list of email objects.", - # No tool-level source_integrity needed - labels are per-item in additional_properties + additional_properties={ + # The local inbox connector is trusted to classify internal senders. + # Per-item labels below may downgrade external messages to UNTRUSTED. + "source_integrity": "trusted", + }, ) async def fetch_emails( count: int = Field(default=5, description="Number of emails to fetch"), @@ -217,8 +221,8 @@ async def fetch_emails( """ emails = SAMPLE_EMAILS[:count] - # Return emails as list[Content] with per-item security labels in additional_properties. - # This ensures FunctionTool.invoke() preserves per-item labels for tier-1 propagation. + # Per-item labels are restriction-only. External messages downgrade the + # locally trusted fallback; internal messages preserve it. result: list[Content] = [] for email in emails: email_text = json.dumps({ diff --git a/python/samples/02-agents/security/user_identity_security_example.py b/python/samples/02-agents/security/user_identity_security_example.py new file mode 100644 index 00000000000..4ccc71d2c6d --- /dev/null +++ b/python/samples/02-agents/security/user_identity_security_example.py @@ -0,0 +1,172 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""USER_IDENTITY principal binding with SecureAgentConfig. + +This sample shows how an application binds identity-scoped data to an +authenticated tenant/user principal and limits destinations to an authorized +principal set. It demonstrates: + +1. Building source and destination tools from host-authenticated identity. +2. Declaring canonical principals with ``PRINCIPAL_METADATA_KEY``. +3. Adding ``SecureAgentConfig`` through ``context_providers``. +4. Allowing a same-principal flow and blocking a cross-principal flow. + +Environment variables: + FOUNDRY_PROJECT_ENDPOINT: Microsoft Foundry project endpoint. + FOUNDRY_MODEL: Model deployment name. + +Before running: + az login + +Run from the ``python`` directory: + uv run samples/02-agents/security/user_identity_security_example.py + +Expected behavior: + - Saving Alice's profile to Alice's destination is allowed. + - Sending Alice's profile to Bob's destination is blocked and audited. +""" + +import asyncio +import os +from typing import Any + +from agent_framework import Agent, AgentSession, FunctionTool, tool +from agent_framework.foundry import FoundryChatClient +from agent_framework.security import PRINCIPAL_METADATA_KEY, SecureAgentConfig +from azure.identity import AzureCliCredential +from dotenv import load_dotenv + +load_dotenv() + +TENANT_ID = "tenant-contoso" +AUTHENTICATED_USER_ID = "alice" +OTHER_USER_ID = "bob" + + +def principal_set(*, tenant_id: str, user_id: str) -> list[dict[str, str]]: + """Return the canonical principal-set representation used by FIDES.""" + return [{"tenant_id": tenant_id, "user_id": user_id}] + + +def create_identity_tools(*, tenant_id: str, authenticated_user_id: str) -> list[FunctionTool]: + """Create tools whose security metadata comes from authenticated host state. + + In a real application, derive these values from the authenticated request or + session before constructing the agent's tools. Never accept the principal + from model-generated tool arguments or remote result metadata. + """ + authenticated_principals = principal_set(tenant_id=tenant_id, user_id=authenticated_user_id) + other_principals = principal_set(tenant_id=tenant_id, user_id=OTHER_USER_ID) + + @tool( + description="Read the authenticated user's profile.", + additional_properties={ + "source_integrity": "trusted", + "confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: authenticated_principals, + }, + ) + async def read_my_profile() -> str: + """Return identity-scoped data for the authenticated user.""" + return f"{authenticated_user_id} profile: preferred language is Python." + + @tool( + description="Save a note to the authenticated user's profile.", + additional_properties={ + "max_allowed_confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: authenticated_principals, + }, + ) + async def save_to_my_profile(note: str) -> dict[str, Any]: + """Save data to a destination authorized for the same principal.""" + print(f"ALLOWED: saved to {authenticated_user_id}: {note}") + return {"status": "saved", "user_id": authenticated_user_id} + + @tool( + description="Send a note to another user's account.", + additional_properties={ + "max_allowed_confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: other_principals, + }, + ) + async def send_to_other_account(note: str) -> dict[str, Any]: + """Represent a destination authorized for a different principal.""" + print(f"UNEXPECTED: sent to {OTHER_USER_ID}: {note}") + return {"status": "sent", "user_id": OTHER_USER_ID} + + return [read_my_profile, save_to_my_profile, send_to_other_account] + + +def create_agent() -> tuple[Agent, SecureAgentConfig]: + """Create a Foundry agent with identity-aware security enforcement.""" + client = FoundryChatClient( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + model=os.environ["FOUNDRY_MODEL"], + credential=AzureCliCredential(), + ) + security = SecureAgentConfig( + auto_hide_untrusted=True, + enable_policy_enforcement=True, + block_on_violation=True, + ) + agent = Agent( + client=client, + name="IdentityScopedAssistant", + instructions="Follow the user's requested tool sequence exactly.", + tools=create_identity_tools( + tenant_id=TENANT_ID, + authenticated_user_id=AUTHENTICATED_USER_ID, + ), + context_providers=[security], + ) + return agent, security + + +async def run_scenario( + agent: Agent, + security: SecureAgentConfig, + *, + title: str, + prompt: str, +) -> None: + """Run one isolated scenario and print policy audit entries.""" + print(f"\n{'=' * 72}\n{title}\n{'=' * 72}") + session = AgentSession() + response = await agent.run(prompt, session=session) + print(f"\nAgent response: {response.text}") + + audit_log = security.get_audit_log(session) + if audit_log: + print("\nPolicy audit:") + for entry in audit_log: + print(f"- {entry.get('reason', 'Policy violation')}") + else: + print("\nPolicy audit: no violations") + + +async def main() -> None: + """Run matching-principal and cross-principal flows.""" + agent, security = create_agent() + + await run_scenario( + agent, + security, + title="Allowed: Alice data to Alice destination", + prompt=( + "Call read_my_profile. Then call save_to_my_profile with the profile text " + "as the note. Do not call any other tools." + ), + ) + await run_scenario( + agent, + security, + title="Blocked: Alice data to Bob destination", + prompt=( + "Call read_my_profile. Then call send_to_other_account with the profile text " + "as the note. Do not call any other tools." + ), + ) + + +if __name__ == "__main__": + asyncio.run(main()) From 1779db9da12b1d40e692e157a0ebaf8c9d38b8f6 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 10 Sep 2026 16:25:00 +0200 Subject: [PATCH 3/3] Python: address security review feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../packages/core/agent_framework/security.py | 16 +-- .../tests/core/test_harness_tool_approval.py | 112 ++++++++++++++++++ python/packages/core/tests/test_security.py | 29 ++++- .../security/FIDES_DEVELOPER_GUIDE.md | 50 ++++++++ 4 files changed, 195 insertions(+), 12 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index 85a2cf07242..5b88738d9a4 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -100,14 +100,14 @@ def _get_additional_properties(obj: Any) -> dict[str, Any]: def _canonical_principals(value: Any, *, source: str) -> tuple[tuple[str, str], ...]: """Validate and canonicalize a principal-set declaration.""" - if not isinstance(value, list) or not value: - raise ValueError(f"{source} principals must be a non-empty list") + if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)) or not value: + raise ValueError(f"{source} principals must be a non-empty sequence") principals: set[tuple[str, str]] = set() - for item in cast(list[Any], value): - if not isinstance(item, dict): + for item in cast(Sequence[Any], value): + if not isinstance(item, Mapping): raise ValueError(f"{source} principals must contain mappings") - principal = cast(dict[str, Any], item) + principal = cast(Mapping[str, Any], item) if set(principal) != {"tenant_id", "user_id"}: raise ValueError(f"{source} principal fields must be tenant_id and user_id") tenant_id = principal.get("tenant_id") @@ -1861,11 +1861,7 @@ def _label_result( old_conf.value, result_label.confidentiality.value, ) - logger.debug( - "Hidden result security metadata merged for '%s': %s", - function_name, - result_label.metadata, - ) + logger.debug("Hidden result security metadata merged for '%s'", function_name) else: logger.info( f"Result from '{function_name}' fully hidden - context label " diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index c36eacc1855..7f6d61d8056 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -35,6 +35,8 @@ ) from agent_framework._feature_stage import ExperimentalWarning from agent_framework.security import ( + PRINCIPAL_METADATA_KEY, + ConfidentialityLabel, ContentLabel, IntegrityLabel, LabelTrackingFunctionMiddleware, @@ -494,6 +496,116 @@ async def capture_response( assert not any(content.type.startswith("function_approval_") for content in model_contents) +@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"]) +async def test_principal_change_requires_visible_second_approval( + chat_client_base: MockBaseChatClient, + streaming: bool, +) -> None: + """A principal change persists one replacement request before execution.""" + received: list[str] = [] + + def principals(user_id: str) -> list[dict[str, str]]: + return [{"tenant_id": "tenant-a", "user_id": user_id}] + + @tool( + name="identity_sink", + additional_properties={ + "max_allowed_confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: principals("user-c"), + }, + ) + def identity_sink(value: str) -> str: + received.append(value) + return "sent" + + tracker = LabelTrackingFunctionMiddleware() + policy = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + session = AgentSession(session_id=f"principal-change-{streaming}") + tracker._scope_for_session(session).context_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata={PRINCIPAL_METADATA_KEY: principals("user-a")}, + ) + agent = Agent( + client=chat_client_base, + tools=[identity_sink], + middleware=[tracker, policy], + context_providers=[InMemoryHistoryProvider()], + ) + function_call = Content.from_function_call( + call_id="principal-call", + name="identity_sink", + arguments={"value": "payload"}, + id="principal-occurrence", + ) + + if streaming: + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[function_call])], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])], + ] + first_stream = agent.run("send identity data", stream=True, session=session) + _ = [update async for update in first_stream] + first = await first_stream.get_final_response() + else: + chat_client_base.run_responses = [ + ChatResponse(messages=Message(role="assistant", contents=[function_call])), + ChatResponse(messages=Message(role="assistant", contents=["done"])), + ] + first = await agent.run("send identity data", session=session) + + original_request = first.user_input_requests[0] + tracker._scope_for_session(session).context_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata={PRINCIPAL_METADATA_KEY: principals("user-b")}, + ) + + if streaming: + stale_stream = agent.run( + original_request.to_function_approval_response(True), + stream=True, + session=session, + ) + stale_updates = [update async for update in stale_stream] + stale = await stale_stream.get_final_response() + assert [(update.role, [content.type for content in update.contents]) for update in stale_updates] == [ + ("assistant", ["function_approval_request"]), + ] + else: + stale = await agent.run(original_request.to_function_approval_response(True), session=session) + + assert received == [] + assert chat_client_base.call_count == 1 + replacement = stale.user_input_requests[0] + assert replacement.id != original_request.id + assert replacement.function_call is not None + assert original_request.function_call is not None + assert replacement.function_call.id == original_request.function_call.id + pending = session.state["tool_approval"]["pending_approval_requests"] + assert [snapshot["id"] for snapshot in pending] == [replacement.id] + + if streaming: + approved_stream = agent.run( + replacement.to_function_approval_response(True), + stream=True, + session=session, + ) + approved_updates = [update async for update in approved_stream] + approved = await approved_stream.get_final_response() + assert [(update.role, [content.type for content in update.contents]) for update in approved_updates] == [ + ("tool", ["function_result"]), + ("assistant", ["text"]), + ] + else: + approved = await agent.run(replacement.to_function_approval_response(True), session=session) + + assert received == ["payload"] + assert chat_client_base.call_count == 2 + assert [(message.role, [content.type for content in message.contents]) for message in approved.messages] == [ + ("tool", ["function_result"]), + ("assistant", ["text"]), + ] + + async def test_replacement_approval_preserves_unanswered_reused_call_id_sibling( chat_client_base: MockBaseChatClient, ) -> None: diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 3c45f2baad6..b4e057118f9 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -6,7 +6,7 @@ import json import logging from datetime import timedelta -from types import SimpleNamespace +from types import MappingProxyType, SimpleNamespace from typing import Any, cast from unittest.mock import AsyncMock @@ -4615,7 +4615,10 @@ async def next_fn() -> None: assert middleware.get_context_label().integrity == IntegrityLabel.TRUSTED assert middleware.get_context_label().confidentiality == ConfidentialityLabel.USER_IDENTITY - async def test_hidden_user_identity_result_unions_principals_at_same_rank(self, middleware, mock_function) -> None: + async def test_hidden_user_identity_result_unions_principals_at_same_rank( + self, middleware, mock_function, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.DEBUG, logger="agent_framework.security") middleware._context_label = ContentLabel( confidentiality=ConfidentialityLabel.USER_IDENTITY, metadata=_principal_metadata("user-a"), @@ -4638,6 +4641,10 @@ async def next_fn() -> None: {"tenant_id": "tenant-a", "user_id": "user-b"}, ] + assert "user-a" not in caplog.text + assert "user-b" not in caplog.text + caplog.clear() + destination = _identity_destination(_principal_metadata("user-a")[_PRINCIPALS_KEY]) policy_context = FunctionInvocationContext(function=destination, arguments={}) policy_context.metadata["context_label"] = middleware.get_context_label() @@ -5467,6 +5474,24 @@ def test_matching_user_identity_to_user_identity_allowed(self): is True ) + def test_user_identity_accepts_sequence_and_mapping_implementations(self) -> None: + from agent_framework.security import check_confidentiality_allowed + + principal = MappingProxyType({"tenant_id": "tenant-a", "user_id": "user-a"}) + ui_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata={_PRINCIPALS_KEY: (principal,)}, + ) + + assert ( + check_confidentiality_allowed( + ui_label, + ConfidentialityLabel.USER_IDENTITY, + authorized_principals=(principal,), + ) + is True + ) + @pytest.mark.parametrize( ("source_metadata", "authorized_principals"), [ diff --git a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md index 8f7efb0dd3d..44f8eefee0f 100644 --- a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md +++ b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md @@ -103,6 +103,56 @@ is blocked. See [`user_identity_security_example.py`](user_identity_security_example.py) for a complete runnable setup. +#### Migrating legacy USER_IDENTITY labels + +Earlier FIDES examples used a single, unnamespaced `user_id` and did not bind +identity destinations. Releases containing principal-bound enforcement reject +that shape. Migrate both the source label and every USER_IDENTITY destination; +do not fill in a tenant or user from model-generated arguments. + +Before: + +```python +legacy_label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata={"user_id": "alice"}, +) + + +@tool( + description="Save identity data", + additional_properties={"max_allowed_confidentiality": "user_identity"}, +) +async def save_identity_data(data: str) -> None: + ... +``` + +After: + +```python +alice = [{"tenant_id": authenticated_tenant_id, "user_id": authenticated_user_id}] + +label = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + metadata={PRINCIPAL_METADATA_KEY: alice}, +) + + +@tool( + description="Save identity data", + additional_properties={ + "max_allowed_confidentiality": "user_identity", + PRINCIPAL_METADATA_KEY: alice, + }, +) +async def save_identity_data(data: str) -> None: + ... +``` + +Combined content may contain more than one principal. A destination must list +all authorized principals because the policy checks that the source set is a +subset of the destination set. + ### 2. Label Tracking Middleware with Tiered Label Propagation `LabelTrackingFunctionMiddleware` uses a **tiered label propagation** scheme where the result label of a tool call is determined by a strict 3-tier priority: