use crate::error::ApiError;
use crate::state::AppState;
use axum::extract::Request;
use notedthat_core::{AccessPolicy, KbSlug, KeyFilter, Principal, Verb};
use std::sync::Arc;
#[derive(Clone)]
pub(crate) struct KbAccess {
kb: KbSlug,
principal: Principal,
policy: Arc<AccessPolicy>,
}
impl KbAccess {
pub(crate) fn resolve(state: &AppState, slug: &str, req: &Request) -> Result<Self, ApiError> {
let kb = crate::router::lookup_kb(state, slug)?;
let policy = state.access_policies.get(slug).cloned().unwrap_or_else(|| {
tracing::error!(
kb = slug,
"declared knowledge base has no access policy; refusing everything"
);
Arc::new(AccessPolicy::empty())
});
Ok(Self {
kb,
principal: crate::middleware::principal(req),
policy,
})
}
pub(crate) fn kb(&self) -> &KbSlug {
&self.kb
}
pub(crate) fn require(&self, verb: Verb, key: &str) -> Result<(), ApiError> {
if self.policy.allows(&self.principal, verb, key) {
return Ok(());
}
Err(self.denial())
}
pub(crate) fn require_any(&self, verb: Verb) -> Result<(), ApiError> {
if self.policy.grants_any(&self.principal, verb) {
return Ok(());
}
Err(self.denial())
}
pub(crate) fn policy_grants_any(&self, verb: Verb) -> bool {
self.policy.grants_any(&self.principal, verb)
}
pub(crate) fn allows(&self, verb: Verb, key: &str) -> bool {
self.policy.allows(&self.principal, verb, key)
}
pub(crate) fn require_visible(&self) -> Result<(), ApiError> {
if self.policy.visible_in_listing(&self.principal) {
return Ok(());
}
Err(self.denial())
}
pub(crate) fn require_service_token(&self) -> Result<(), ApiError> {
if self.principal.is_service_token() {
return Ok(());
}
Err(self.denial())
}
pub(crate) fn filter(&self, verb: Verb) -> KeyFilter<'_> {
self.policy.key_filter(&self.principal, verb)
}
fn denial(&self) -> ApiError {
match self.principal {
Principal::Anyone => crate::router::kb_not_found(self.kb.as_str()),
Principal::SignedIn(_) => ApiError::Forbidden,
}
}
}
pub(crate) enum ScanScope {
From(Option<String>),
Disjoint,
}
pub(crate) fn effective_prefix(requested: Option<&str>, hint: Option<&str>) -> ScanScope {
match (requested, hint) {
(None, None) => ScanScope::From(None),
(Some(prefix), None) => ScanScope::From(Some(prefix.to_string())),
(None, Some(hint)) => ScanScope::From(Some(hint.to_string())),
(Some(prefix), Some(hint)) if prefix.starts_with(&format!("{hint}/")) => {
ScanScope::From(Some(prefix.to_string()))
}
(Some(prefix), Some(hint)) if hint.starts_with(prefix) => {
ScanScope::From(Some(hint.to_string()))
}
(Some(_), Some(_)) => ScanScope::Disjoint,
}
}
pub(crate) fn visible_in_listing(state: &AppState, slug: &str, principal: &Principal) -> bool {
match state.access_policies.get(slug) {
Some(policy) => policy.visible_in_listing(principal),
None => AccessPolicy::empty().visible_in_listing(principal),
}
}
#[cfg(test)]
mod tests {
use super::{ScanScope, effective_prefix};
fn scan(requested: Option<&str>, hint: Option<&str>) -> String {
match effective_prefix(requested, hint) {
ScanScope::From(None) => "<whole knowledge base>".to_string(),
ScanScope::From(Some(prefix)) => prefix,
ScanScope::Disjoint => "<disjoint>".to_string(),
}
}
#[test]
fn an_unscoped_grant_leaves_the_caller_prefix_alone() {
assert_eq!(scan(None, None), "<whole knowledge base>");
assert_eq!(scan(Some("docs/"), None), "docs/");
}
#[test]
fn a_scoped_grant_narrows_a_whole_knowledge_base_scan() {
assert_eq!(scan(None, Some("public")), "public");
}
#[test]
fn the_caller_prefix_wins_when_it_is_inside_the_granted_segment() {
assert_eq!(scan(Some("public/deep/"), Some("public")), "public/deep/");
}
#[test]
fn the_grant_wins_when_the_caller_asks_at_or_above_it() {
assert_eq!(scan(Some("public"), Some("public")), "public");
assert_eq!(scan(Some("pub"), Some("public")), "public");
assert_eq!(scan(Some(""), Some("public")), "public");
}
#[test]
fn a_sibling_sharing_only_a_textual_prefix_is_disjoint() {
assert_eq!(scan(Some("public-internal/"), Some("public")), "<disjoint>");
assert_eq!(scan(Some("publicity.md"), Some("public")), "<disjoint>");
assert_eq!(scan(Some("archive/"), Some("public")), "<disjoint>");
}
}