knafeh 1.1.0

QUIC-based RPC library with Python bindings
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
151
152
153
154

//! Shared test infrastructure — services, TLS certs, server helpers.
//!
//! Used by all E2E test files to avoid duplicating boilerplate.
#![allow(dead_code)]

use std::net::SocketAddr;
use std::sync::LazyLock;

use async_trait::async_trait;
use rcgen::{
    BasicConstraints, CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa, KeyPair,
    KeyUsagePurpose, SanType,
};
use tokio::sync::oneshot;
use tokio::task::JoinHandle;

use knafeh::error::KnafehError;
use knafeh::rpc::message::{RpcRequest, RpcResponse};
use knafeh::rpc::service::{MethodDescriptor, MethodKind, Service};
use knafeh::rpc::stream::{RpcStreamRequest, RpcStreamResponse};
use knafeh::transport::tls::TlsConfig;
use knafeh::Server;

// ---------------------------------------------------------------------------
// Ephemeral TLS certificates
// ---------------------------------------------------------------------------

pub struct TestCerts {
    pub _dir: tempfile::TempDir,
    pub cert_path: String,
    pub key_path: String,
    pub ca_path: String,
}

pub static TEST_CERTS: LazyLock<TestCerts> = LazyLock::new(|| {
    let mut ca_params =
        CertificateParams::new(Vec::new()).expect("empty SAN list is valid for a CA");
    ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
    ca_params
        .distinguished_name
        .push(DnType::CommonName, "Knafeh test CA");
    ca_params.key_usages = vec![
        KeyUsagePurpose::DigitalSignature,
        KeyUsagePurpose::KeyCertSign,
        KeyUsagePurpose::CrlSign,
    ];
    let ca_key = KeyPair::generate().expect("failed to generate test CA key");
    let ca_cert = ca_params
        .self_signed(&ca_key)
        .expect("failed to generate test CA cert");

    let mut server_params = CertificateParams::new(vec!["localhost".to_string()])
        .expect("failed to build test server cert params");
    // quiche verifies the endpoint host as a DNS name, even when tests connect
    // to the IPv4 loopback literal.
    server_params
        .subject_alt_names
        .push(SanType::DnsName("127.0.0.1".try_into().unwrap()));
    server_params
        .distinguished_name
        .push(DnType::CommonName, "localhost");
    server_params.key_usages = vec![
        KeyUsagePurpose::DigitalSignature,
        KeyUsagePurpose::KeyEncipherment,
    ];
    server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
    server_params.use_authority_key_identifier_extension = true;
    let server_key = KeyPair::generate().expect("failed to generate test server key");
    let server_cert = server_params
        .signed_by(&server_key, &ca_cert, &ca_key)
        .expect("failed to generate test server cert");

    let dir = tempfile::tempdir().expect("failed to create temp dir");
    let cert_path = dir.path().join("cert.pem");
    let key_path = dir.path().join("key.pem");
    let ca_path = dir.path().join("ca.pem");
    std::fs::write(&cert_path, server_cert.pem()).unwrap();
    std::fs::write(&key_path, server_key.serialize_pem()).unwrap();
    std::fs::write(&ca_path, ca_cert.pem()).unwrap();
    TestCerts {
        _dir: dir,
        cert_path: cert_path.to_string_lossy().to_string(),
        key_path: key_path.to_string_lossy().to_string(),
        ca_path: ca_path.to_string_lossy().to_string(),
    }
});

pub fn tls_server() -> TlsConfig {
    TlsConfig::server(&TEST_CERTS.cert_path, &TEST_CERTS.key_path)
}

// ---------------------------------------------------------------------------
// Server helper
// ---------------------------------------------------------------------------

pub async fn start_test_server(server: Server) -> (SocketAddr, JoinHandle<()>) {
    let (ready_tx, ready_rx) = oneshot::channel();
    let handle = tokio::spawn(async move {
        server.serve_with_ready_signal(ready_tx).await.unwrap();
    });
    (ready_rx.await.unwrap(), handle)
}

// ---------------------------------------------------------------------------
// Echo service (used everywhere)
// ---------------------------------------------------------------------------

pub struct EchoService;

#[async_trait]
impl Service for EchoService {
    fn name(&self) -> &str {
        "echo"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "echo".to_string(),
            kind: MethodKind::Unary,
        }]
    }

    async fn call_unary(
        &self,
        _method: &str,
        request: RpcRequest,
    ) -> Result<RpcResponse, KnafehError> {
        Ok(RpcResponse::ok(request.body))
    }

    async fn call_server_stream(
        &self,
        _: &str,
        _: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}