apeirogond 0.15.2

Daemon for Apeirogon, a cross-platform software KVM
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

// SPDX-License-Identifier: GPL-3.0-or-later

use std::net::SocketAddr;
use std::time::Instant;

use a7n_core::pairing::{Paired, PairingCode, RequesterPairing};
use a7n_proto::v1::PairDone;
use quinn::Connection;
use rand_core::OsRng;
use tokio::sync::{OwnedSemaphorePermit, oneshot};

use super::{Daemon, DaemonError, Event, unix_seconds};
use crate::command::Reply;
use crate::pairing::{self, PairingError};
use crate::pin_store::Entry;
use crate::tls;

impl Daemon {
    pub(super) fn host_pairing(
        &mut self,
        connection: Connection,
        permit: OwnedSemaphorePermit,
    ) -> Result<(), DaemonError> {
        // Refused or rate-limited attempts must not close the window, or any
        // LAN host could cancel pairing for everyone.
        let open = self
            .trust
            .read()
            .map_err(|_| DaemonError::TrustPoisoned)?
            .pairing_open();
        let allowed = open
            && self.host.is_some()
            && !self.host_busy
            && self
                .limiter
                .allow(connection.remote_address().ip(), Instant::now());
        let Some(host) = allowed.then(|| self.host.take()).flatten() else {
            connection.close(crate::link::CLOSE_NORMAL, b"");
            return Ok(());
        };
        self.trust
            .write()
            .map_err(|_| DaemonError::TrustPoisoned)?
            .pairing_until = None;
        self.host_busy = true;
        let info = self.info.clone();
        let events = self.events.clone();
        self.tasks.spawn(async move {
            let admission_events = events.clone();
            let result = pairing::respond(
                connection,
                host,
                info,
                move |entry, paired| async move {
                    let (reply, receiver) = oneshot::channel();
                    admission_events
                        .send(Event::Confirmed {
                            entry,
                            paired,
                            reply,
                        })
                        .await
                        .map_err(|_| PairingError::Refused)?;
                    receiver.await.map_err(|_| PairingError::Refused)?
                },
                Instant::now,
                &mut OsRng,
            )
            .await;
            drop(permit);
            if result.is_err() {
                crate::log::event!("kind=pairing_failed");
            }
            let _ = events.send(Event::PairingEnded).await;
        });
        Ok(())
    }

    pub(super) fn admit(
        &mut self,
        entry: &Entry,
        paired: Paired,
    ) -> Result<PairDone, PairingError> {
        let result = unix_seconds().and_then(|now| {
            self.desk
                .admit(&self.identity, &mut self.pins, entry, now)?;
            Ok(())
        });
        let trust = self.sync_trust();
        self.drop_untrusted();
        self.broadcast(&a7n_proto::v1::Envelope {
            body: Some(a7n_proto::v1::envelope::Body::Desk(self.desk.to_desk())),
        });
        if result.is_err() || trust.is_err() {
            crate::log::event!("kind=pairing_store_failed");
            return Err(PairingError::Refused);
        }
        Ok(paired.done())
    }

    pub(super) fn join(&mut self, addr: SocketAddr, code: PairingCode, reply: Reply<Entry>) {
        let config = match tls::pair_client_config(&self.identity) {
            Ok(config) => config,
            Err(error) => {
                let _ = reply.send(Err(error.into()));
                return;
            }
        };
        let mut endpoint = self.endpoint.clone();
        endpoint.set_default_client_config(config);
        let requester =
            RequesterPairing::new(code, None, self.identity.pin, self.info.clone(), &mut OsRng);
        let events = self.events.clone();
        self.tasks.spawn(async move {
            let result = async {
                let connection = pairing::dial(&endpoint, addr).await?;
                pairing::initiate(connection, requester).await
            }
            .await;
            let _ = events
                .send(Event::Joined {
                    result,
                    addr,
                    reply,
                })
                .await;
        });
    }

    pub(super) fn store_peer(&mut self, entry: Entry) -> Result<(), DaemonError> {
        let mut staged = self.pins.clone();
        staged.add(entry);
        staged.save()?;
        self.pins = staged;
        self.sync_trust()
    }
}