systemprompt-security 0.63.1

//! Gate admin-only tools by [`AccessScope`]; registered as the `scope_check`
//! policy.
//!
//! Reads the typed `ctx.access_scope` the enforcement point resolved (agent
//! YAML `oauth.scopes`, JWT permissions, DB roles). Configurable via:
//!
//! ```yaml
//! - id: scope_check
//!   admin_only_prefixes:
//!     - "mcp__systemprompt__"
//! ```
//!
//! Copyright (c) systemprompt.io — Business Source License 1.1.
//! See <https://systemprompt.io> for licensing details.

use std::borrow::Cow;

use serde::Deserialize;
use serde_yaml::Value as YamlValue;
use systemprompt_identifiers::PolicyId;

use super::super::registry::{PolicyConfigurationError, PolicyRegistration};
use super::super::types::{AccessScope, GovernancePolicy, PolicyContext};
use crate::authz::types::{Decision, DenyReason, MatchedBy};

const ID: &str = "scope_check";
const DEFAULT_ADMIN_ONLY_PREFIXES: &[&str] = &["mcp__systemprompt__"];

#[derive(Debug)]
struct ScopeCheck {
    admin_only_prefixes: Vec<String>,
}

#[derive(Debug, Default, Deserialize)]
struct ScopeCheckYaml {
    #[serde(default)]
    admin_only_prefixes: Vec<String>,
}

impl ScopeCheck {
    fn from_yaml(v: &YamlValue) -> Result<Self, PolicyConfigurationError> {
        let cfg = serde_yaml::from_value::<Option<ScopeCheckYaml>>(v.clone())
            .map_err(|source| PolicyConfigurationError::Yaml {
                context: "malformed scope_check policy entry",
                source,
            })?
            .unwrap_or_default();
        let admin_only_prefixes = if cfg.admin_only_prefixes.is_empty() {
            DEFAULT_ADMIN_ONLY_PREFIXES
                .iter()
                .map(|s| (*s).to_owned())
                .collect()
        } else {
            cfg.admin_only_prefixes
        };
        Ok(Self {
            admin_only_prefixes,
        })
    }
}

impl GovernancePolicy for ScopeCheck {
    fn id(&self) -> PolicyId {
        PolicyId::new(ID)
    }
    fn name(&self) -> &'static str {
        "Scope Check"
    }
    fn description(&self) -> &'static str {
        "Block non-admin agents from calling tools whose name starts with an \
         admin-only prefix (default: mcp__systemprompt__)."
    }
    fn evaluate(&self, ctx: &PolicyContext<'_>) -> Decision {
        if ctx.access_scope == AccessScope::Admin {
            return Decision::Allow {
                matched_by: MatchedBy::PolicyAllow {
                    policy_id: PolicyId::new(ID),
                    detail: Cow::Borrowed("admin scope grants unrestricted tool access"),
                },
            };
        }

        let Some(tool) = ctx.target.tool() else {
            return Decision::Allow {
                matched_by: MatchedBy::PolicyAllow {
                    policy_id: PolicyId::new(ID),
                    detail: Cow::Borrowed("Not a tool call"),
                },
            };
        };
        let tool_str = tool.as_str();
        let requires_admin = self
            .admin_only_prefixes
            .iter()
            .any(|prefix| tool_str.starts_with(prefix.as_str()));

        if requires_admin {
            return Decision::Deny {
                reason: DenyReason::ScopeViolation {
                    tool: tool.clone(),
                    required: AccessScope::Admin,
                },
            };
        }

        let detail = match ctx.access_scope {
            AccessScope::Unknown => {
                Cow::Borrowed("Agent scope could not be resolved; allowed for non-admin tool")
            },
            scope => Cow::Owned(format!("{scope} scope is allowed for tool: {tool_str}")),
        };
        Decision::Allow {
            matched_by: MatchedBy::PolicyAllow {
                policy_id: PolicyId::new(ID),
                detail,
            },
        }
    }
}

inventory::submit! {
    PolicyRegistration {
        id: ID,
        factory: |v| Ok(Box::new(ScopeCheck::from_yaml(v)?)),
    }
}