piying 0.1.0

Fault-tolerant Async Actors Built on Tokio
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

impl Behaviour {
    fn validate_provider_registration(&mut self, provider: &kad::ProviderRecord) -> bool {
        #[cfg_attr(not(feature = "tracing"), allow(unused_variables))]
        let source = &provider.provider;

        // Should be valid UTF8
        let key_str = match std::str::from_utf8(provider.key.as_ref()) {
            Ok(s) => s,
            Err(_) => {
                #[cfg(feature = "tracing")]
                tracing::warn!("invalid UTF-8 in provider key from {source}");
                return false;
            }
        };

        if !self.is_valid_actor_name(key_str) {
            #[cfg(feature = "tracing")]
            tracing::warn!("invalid actor name in provider registration from {source}: {key_str}");
            return false;
        }

        if !self.is_valid_provider_record(provider) {
            #[cfg(feature = "tracing")]
            tracing::warn!("invalid provider record from {source}");
            return false;
        }

        #[cfg(feature = "tracing")]
        tracing::debug!("validated provider registration for {key_str} from {source}");
        true
    }

    fn is_valid_provider_record(&self, provider: &kad::ProviderRecord) -> bool {
        if let Some(expires) = provider.expires {
            // Must not be expired
            if expires < Instant::now() {
                #[cfg(feature = "tracing")]
                tracing::warn!("provider record is already expired");
                return false;
            }
        }

        true
    }

    fn validate_metadata_record(&mut self, source: &PeerId, record: &kad::Record) -> bool {
        // Validate key format: "{name}:meta:{peer_id}"
        let key_str = match str::from_utf8(record.key.as_ref()) {
            Ok(s) => s,
            Err(_) => {
                #[cfg(feature = "tracing")]
                tracing::warn!("invalid UTF-8 in metadata record key from {source}");
                return false;
            }
        };

        let parts: Vec<_> = key_str.splitn(2, ":meta:").collect();
        if parts.len() != 2 {
            #[cfg(feature = "tracing")]
            tracing::warn!("invalid metadata key format from {source}: {key_str}");
            return false;
        }

        let actor_name = parts[0];
        let peer_id_str = parts[1];

        if !self.is_valid_actor_name(actor_name) {
            #[cfg(feature = "tracing")]
            tracing::warn!("invalid actor name from {source}: {actor_name}");
            return false;
        }

        // Parse and validate peer ID from key
        let record_peer_id = match PeerId::from_str(peer_id_str) {
            Ok(id) => id,
            Err(_) => {
                #[cfg(feature = "tracing")]
                tracing::warn!("invalid peer ID in metadata key from {source}: {peer_id_str}");
                return false;
            }
        };

        // Verify the peer_id in key matches the source (prevents impersonation)
        if record_peer_id != *source {
            #[cfg(feature = "tracing")]
            tracing::warn!(
                "peer ID mismatch: source {source} trying to register metadata for peer {record_peer_id}"
            );
            return false;
        }

        // Validate metadata format and size
        if !self.is_valid_metadata(&record.value) {
            #[cfg(feature = "tracing")]
            tracing::warn!("invalid metadata format from {source}");
            return false;
        }

        // Disabled for now: what if the record gets received before the start providing message
        // Verify the source peer is actually providing this actor
        // let actor_key = kad::RecordKey::new(&actor_name);
        // let providers = self.kademlia.store_mut().providers(&actor_key);

        // let is_provider = providers
        //     .iter()
        //     .any(|provider_record| provider_record.provider == *source);

        // if !is_provider {
        //     #[cfg(feature = "tracing")]
        //     tracing::warn!(
        //         "peer {source} trying to register metadata for {actor_name} without being a provider"
        //     );
        //     return false;
        // }

        #[cfg(feature = "tracing")]
        tracing::debug!("validated metadata record for {actor_name} from {source}");

        true
    }

    fn is_valid_actor_name(&self, name: &str) -> bool {
        !name.is_empty()
    }

    fn is_valid_metadata(&self, data: &[u8]) -> bool {
        // Size check
        if data.len() > 64 * 1024 {
            // 64KB limit
            return false;
        }

        ActorRegistration::from_bytes(data).is_ok()
    }
}