prns-runtime-tokio 0.3.6

Tokio host runtime for Personal Reticulum
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
140
141
142
143
144
145
146
147
148
149
150

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use tokio::sync::mpsc;

use crate::identity::IdentityHash;
use crate::routing::links::resources::{ResourceCompression, ResourceOffer};
use crate::routing::links::LinkId;

use super::PrnsNodeHandle;

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ResourceAdmissionPeer {
    Any,
    Authenticated(IdentityHash),
    AuthenticatedOneOf(Arc<[IdentityHash]>),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResourceOfferAdmission {
    pub peer: ResourceAdmissionPeer,
    pub max_uncompressed_bytes: u64,
    pub accept_compressed: bool,
}

pub struct ResourceOfferMonitor {
    offers: mpsc::UnboundedReceiver<ResourceOffer>,
}

impl ResourceOfferMonitor {
    pub async fn recv(&mut self) -> Option<ResourceOffer> {
        self.offers.recv().await
    }
}

struct AdmissionEntry {
    rule: ResourceOfferAdmission,
    offers: mpsc::UnboundedSender<ResourceOffer>,
}

#[derive(Clone, Default)]
pub(crate) struct ResourceAdmissionRegistry {
    entries: Arc<Mutex<HashMap<LinkId, AdmissionEntry>>>,
}

impl ResourceAdmissionRegistry {
    pub(crate) fn install(
        &self,
        link_id: LinkId,
        rule: ResourceOfferAdmission,
    ) -> ResourceOfferMonitor {
        let (offers, receiver) = mpsc::unbounded_channel();
        if let Ok(mut entries) = self.entries.lock() {
            entries.insert(link_id, AdmissionEntry { rule, offers });
        }
        ResourceOfferMonitor { offers: receiver }
    }

    pub(crate) fn remove(&self, link_id: LinkId) {
        if let Ok(mut entries) = self.entries.lock() {
            entries.remove(&link_id);
        }
    }

    pub(crate) fn permits(&self, offer: &ResourceOffer) -> bool {
        let Ok(entries) = self.entries.lock() else {
            return false;
        };
        let Some(entry) = entries.get(&offer.link_id) else {
            return false;
        };
        if offer.uncompressed_data_bytes > entry.rule.max_uncompressed_bytes {
            return false;
        }
        if !entry.rule.accept_compressed && offer.compression == ResourceCompression::Bz2 {
            return false;
        }
        let peer_allowed = match &entry.rule.peer {
            ResourceAdmissionPeer::Any => true,
            ResourceAdmissionPeer::Authenticated(expected) => {
                offer.remote_identity == Some(*expected)
            }
            ResourceAdmissionPeer::AuthenticatedOneOf(expected) => offer
                .remote_identity
                .is_some_and(|identity| expected.contains(&identity)),
        };
        if peer_allowed {
            let _ = entry.offers.send(*offer);
        }
        peer_allowed
    }
}

impl PrnsNodeHandle {
    pub fn admit_resource_offers(
        &self,
        link_id: LinkId,
        admission: ResourceOfferAdmission,
    ) -> ResourceOfferMonitor {
        self.resource_admission.install(link_id, admission)
    }

    pub fn deny_resource_offers(&self, link_id: LinkId) {
        self.resource_admission.remove(link_id);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::routing::links::resources::{ResourceHash, ResourceOffer};

    const LINK: LinkId = LinkId::new([0x11; 16]);
    const PEER: IdentityHash = IdentityHash::new([0x22; 16]);

    fn offer(identity: Option<IdentityHash>) -> ResourceOffer {
        ResourceOffer {
            link_id: LINK,
            remote_identity: identity,
            hash: ResourceHash::new([0x33; 32]),
            uncompressed_data_bytes: 1024,
            sealed_transfer_bytes: 900,
            part_count: 2,
            segment_index: 1,
            total_segment_count: 1,
            compression: ResourceCompression::Bz2,
            has_metadata: true,
        }
    }

    #[tokio::test]
    async fn authenticated_admission_is_per_link_identity_size_and_compression() {
        let registry = ResourceAdmissionRegistry::default();
        let mut monitor = registry.install(
            LINK,
            ResourceOfferAdmission {
                peer: ResourceAdmissionPeer::Authenticated(PEER),
                max_uncompressed_bytes: 2048,
                accept_compressed: true,
            },
        );
        assert!(!registry.permits(&offer(None)));
        assert!(!registry.permits(&offer(Some(IdentityHash::new([0x44; 16])))));
        let accepted = offer(Some(PEER));
        assert!(registry.permits(&accepted));
        assert_eq!(monitor.recv().await, Some(accepted));
        registry.remove(LINK);
        assert!(!registry.permits(&accepted));
    }
}