canic-core 0.26.0

Canic — a canister orchestration and management toolkit for the Internet Computer
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

use super::dependency_unavailable;
use crate::{
    access::AccessError,
    cdk::{api::msg_arg_data, candid::de::IDLDeserialize, types::Principal},
    dto::auth::DelegatedToken,
    ops::{
        auth::{DelegatedTokenOps, VerifiedDelegatedToken},
        ic::IcOps,
        runtime::env::EnvOps,
    },
};

const MAX_INGRESS_BYTES: usize = 64 * 1024; // 64 KiB

///
/// CallerBoundToken
///
/// Verified delegated token that has passed caller-subject binding.
struct CallerBoundToken {
    verified: VerifiedDelegatedToken,
}

impl CallerBoundToken {
    /// bind_to_caller
    ///
    /// Enforce subject binding and return a caller-bound token wrapper.
    fn bind_to_caller(
        verified: VerifiedDelegatedToken,
        caller: Principal,
    ) -> Result<Self, AccessError> {
        enforce_subject_binding(verified.claims.subject(), caller)?;
        Ok(Self { verified })
    }

    /// scopes
    ///
    /// Borrow token scopes after caller binding has been enforced.
    fn scopes(&self) -> &[String] {
        self.verified.claims.scopes()
    }

    /// into_verified
    ///
    /// Unwrap the verified delegated token for downstream consumers.
    fn into_verified(self) -> VerifiedDelegatedToken {
        self.verified
    }
}

pub(super) async fn delegated_token_verified(
    authenticated_subject: Principal,
    required_scope: Option<&str>,
) -> Result<VerifiedDelegatedToken, AccessError> {
    let token = delegated_token_from_args()?;

    let authority_pid =
        EnvOps::root_pid().map_err(|_| dependency_unavailable("root pid unavailable"))?;

    let now_secs = IcOps::now_secs();
    let self_pid = IcOps::canister_self();

    verify_token(
        token,
        authenticated_subject,
        authority_pid,
        now_secs,
        self_pid,
        required_scope,
    )
    .await
}

/// Verify a delegated token against the configured authority.
#[expect(clippy::unused_async)]
async fn verify_token(
    token: DelegatedToken,
    caller: Principal,
    authority_pid: Principal,
    now_secs: u64,
    self_pid: Principal,
    required_scope: Option<&str>,
) -> Result<VerifiedDelegatedToken, AccessError> {
    let verified = DelegatedTokenOps::verify_token(&token, authority_pid, now_secs, self_pid)
        .map_err(|err| AccessError::Denied(err.to_string()))?;

    let caller_bound = CallerBoundToken::bind_to_caller(verified, caller)?;
    enforce_required_scope(required_scope, caller_bound.scopes())?;

    Ok(caller_bound.into_verified())
}

pub(super) fn enforce_subject_binding(
    sub: Principal,
    caller: Principal,
) -> Result<(), AccessError> {
    if sub == caller {
        Ok(())
    } else {
        Err(AccessError::Denied(format!(
            "delegated token subject '{sub}' does not match caller '{caller}'"
        )))
    }
}

pub(super) fn enforce_required_scope(
    required_scope: Option<&str>,
    token_scopes: &[String],
) -> Result<(), AccessError> {
    let Some(required_scope) = required_scope else {
        return Ok(());
    };

    if token_scopes.iter().any(|scope| scope == required_scope) {
        Ok(())
    } else {
        Err(AccessError::Denied(format!(
            "delegated token missing required scope '{required_scope}'"
        )))
    }
}

fn delegated_token_from_args() -> Result<DelegatedToken, AccessError> {
    let bytes = msg_arg_data();

    if bytes.len() > MAX_INGRESS_BYTES {
        return Err(AccessError::Denied(
            "delegated token payload exceeds size limit".to_string(),
        ));
    }

    let mut decoder = IDLDeserialize::new(&bytes)
        .map_err(|err| AccessError::Denied(format!("failed to decode ingress arguments: {err}")))?;

    decoder.get_value::<DelegatedToken>().map_err(|err| {
        AccessError::Denied(format!(
            "failed to decode delegated token as first argument: {err}"
        ))
    })
}