holger_ron/exposed/
http2_backend.rs

1use http_body_util::BodyExt;
2use std::{
3    fs::File,
4    io::BufReader,
5    path::Path,
6};
7
8use anyhow::Context;
9use hyper::{
10    body::Incoming as Body,
11    service::service_fn,
12};
13use rustls_pemfile::certs;
14use tokio::net::TcpListener;
15use tokio_rustls::TlsAcceptor;
16use tokio_rustls::rustls::{pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs1KeyDer}, ServerConfig};
17
18use std::sync::Arc;
19use std::sync::atomic::{AtomicBool, Ordering};
20use anyhow::Result;
21use bytes::Bytes;
22use http_body_util::combinators::BoxBody;
23use http_body_util::Full;
24use hyper::{Request, Response, StatusCode};
25use hyper::server::conn::http2;
26use hyper_util::rt::TokioIo;
27
28/// HTTP2 backend holding routing to repository backends
29#[derive(Clone)]
30pub struct Http2Backend {
31    pub name: String,
32    pub listener_addr: String,
33//    pub tls_config: Arc<ServerConfig>,
34    pub tls_config: Option<Arc<ServerConfig>>,
35    pub running: Arc<AtomicBool>,
36    pub fast_routes: Option<FastRoutes>,
37}
38
39// ✅ Implement the real Default trait
40impl Default for Http2Backend {
41    fn default() -> Self {
42        Self::newd()
43    }
44}
45
46impl Http2Backend {
47    /// Register a repository backend to a sub-URL
48    fn newd() -> Self {
49        Http2Backend {
50
51            // init fields
52            name: "".to_string(),
53            listener_addr: "".to_string(),
54            tls_config: None,
55            running: Arc::new(Default::default()),
56            fast_routes: None,
57        }
58    }
59
60    pub fn new(name: String, listener_addr: String,tls_config: Arc<ServerConfig>) -> Self {
61        Self {
62            name,
63            listener_addr,
64            tls_config: Some(tls_config),
65            running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
66            fast_routes: None,
67        }
68    }
69    /// Inject FastRoutes after construction (2‑pass wiring)
70
71    pub(crate) fn set_fast_routes(&mut self, routes: FastRoutes) {
72        self.fast_routes = Some(routes);
73    }
74
75
76
77    pub fn parse_ip_port(url: &str) -> (String, u16) {
78        let clean = url.trim_end_matches('/');
79        let without_scheme = clean.split("://").nth(1).unwrap_or(clean);
80        let mut parts = without_scheme.split(':');
81        let ip = parts.next().unwrap_or("127.0.0.1").to_string();
82        let port = parts.next().and_then(|p| p.parse().ok()).unwrap_or(443);
83        (ip, port)
84    }
85    pub fn backend_from_config(ep: &ExposedEndpoint) -> anyhow::Result<Self> {
86        let tls_config = Arc::new(load_tls_config(ep.ron_cert.as_str(), ep.ron_key.as_str())?);
87
88
89        let (host, port) = Self::parse_ip_port(ep.ron_url.as_str());
90        let listener_addr = format!("{}:{}", host, port);
91
92        Ok(Self {
93            name: ep.ron_name.clone(),
94            listener_addr,
95            tls_config: Some(tls_config),
96            running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
97            fast_routes: None, // routes attached later
98        })
99    }
100
101    /// ✅ Now takes &self, safe to call from factory
102    /// Start serving requests asynchronously (spawns a background task)
103    /// 2️⃣ Start serving HTTPS + HTTP/2 using the internal routing map
104    pub async fn start(self: Arc<Self>) -> anyhow::Result<JoinHandle<()>> {
105        let listener = TcpListener::bind(&self.listener_addr).await?;
106        let tls_acceptor = self.tls_config.as_ref().cloned().map(TlsAcceptor::from);
107//        let tls_acceptor = TlsAcceptor::from(self.tls_config.clone());
108        self.running.store(true, Ordering::SeqCst);
109
110        println!("Listening on https://{}", self.listener_addr);
111
112        let running = self.running.clone();
113        let handle = tokio::spawn(async move {
114            loop {
115                if !running.load(Ordering::SeqCst) {
116                    println!("Http2Backend stopped");
117                    break;
118                }
119
120                let Ok((stream, _)) = listener.accept().await else {
121                    eprintln!("TCP accept failed");
122                    continue;
123                };
124
125                let acceptor = tls_acceptor.clone();
126
127
128                let this = Arc::clone(&self);
129
130                tokio::spawn(async move {
131                    let Some(acceptor) = acceptor else {
132                        eprintln!("No TLS acceptor configured");
133                        return;
134                    };
135
136                    let Ok(tls_stream) = acceptor.accept(stream).await else {
137                        eprintln!("TLS handshake failed");
138                        return;
139                    };
140                    let io = TokioIo::new(tls_stream);
141                    let builder = http2::Builder::new(hyper_util::rt::TokioExecutor::new());
142
143                    if let Err(err) = builder
144                        .serve_connection(io, {
145                            let this = Arc::clone(&this); // one clone per connection
146                            service_fn(move |req| {
147                                // Only clone Arc if needed per request
148                                let this = Arc::clone(&this);
149                                async move { this.handle_request(req).await }
150                            })
151                        })
152                        .await
153                    {
154                        eprintln!("Connection error: {:?}", err);
155                    }
156                });
157            }
158        });
159
160        Ok(handle)
161    }
162
163    pub(crate) fn stop(&self) -> anyhow::Result<()> {
164        // Signal the server loop to exit
165        self.running.store(false, std::sync::atomic::Ordering::SeqCst);
166        println!("Stopping HTTP2 backend on {}", self.listener_addr);
167        Ok(())
168    }
169
170
171//    async fn handle_request(self: Arc<Self>, req: Request<Body>) -> Result<Response<BoxBody<Bytes, Infallible>>, hyper::Error>
172    async fn handle_request(
173        self: Arc<Self>,
174        req: Request<Body>,
175    ) -> Result<Response<BoxBody<Bytes, std::convert::Infallible>>, hyper::Error> {
176        let path = req.uri().path().to_string();
177        let suburl = path.trim_start_matches('/');
178
179        let body_bytes = req.into_body().collect().await?.to_bytes();
180        let body_vec = body_bytes.to_vec();
181        let repo_key = suburl.split('/').next().unwrap_or("");
182        println!("Repo key: {}", repo_key);
183        if let Some(repo) = self.fast_routes.as_ref().and_then(|routes| routes.lookup(repo_key)) {
184            println!("routing to repo.name={}", repo.name());
185
186
187
188
189            let suburl_owned = suburl.to_string();
190            let body_owned = body_vec;
191            let repo_clone = Arc::clone(repo);
192
193            let result = tokio::task::spawn_blocking(move || {
194                repo_clone.handle_http2_request(&suburl_owned, &body_owned)
195            })
196                .await
197                .unwrap();
198
199            match result {
200                Ok((status, headers, data)) => {
201                    let mut response = Response::builder()
202                        .status(StatusCode::from_u16(status).unwrap());
203                    for (k, v) in headers {
204                        response = response.header(k, v);
205                    }
206                    Ok(response.body(Full::new(Bytes::from(data)).boxed()).unwrap())
207                }
208                Err(_) => Ok(Response::builder()
209                    .status(StatusCode::INTERNAL_SERVER_ERROR)
210                    .body(Full::new(Bytes::from_static(b"Internal Error")).boxed())
211                    .unwrap()),
212            }
213        } else {
214            Ok(Response::builder()
215                .status(StatusCode::NOT_FOUND)
216                .body(Full::new(Bytes::from_static(b"Repo Not Found")).boxed())
217                .unwrap())
218        }
219    }
220
221
222}
223
224
225
226fn load_tls_config(cert_path: &str, key_path: &str) -> Result<ServerConfig> {
227    println!("{}", std::env::current_dir()?.display());
228    let certs = load_certs(cert_path)?;
229    let key = load_key(Path::new(key_path))?;
230
231    let mut config = ServerConfig::builder()
232        .with_no_client_auth()
233        .with_single_cert(certs, key)?;
234
235    // Enable HTTP/2 ALPN
236    config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
237
238    Ok(config)
239}
240
241fn load_certs(path: &str) -> Result<Vec<CertificateDer<'static>>> {
242    let mut file = BufReader::new(File::open(path)?);
243    let certs = certs(&mut file)
244        .context("failed to read certificates")?
245        .into_iter()
246        .map(CertificateDer::from)
247        .collect();
248    Ok(certs)
249}
250
251
252use rustls_pemfile::{read_all, Item};
253use tokio::task::JoinHandle;
254use tokio_rustls::rustls::pki_types::{PrivatePkcs8KeyDer, PrivateSec1KeyDer};
255use holger_traits::RepositoryBackendTrait;
256use crate::exposed::ExposedEndpoint;
257use crate::exposed::fast_routes::FastRoutes;
258use crate::Repository;
259
260fn load_key(path: &Path) -> Result<PrivateKeyDer<'static>> {
261    let file = File::open(path)?;
262
263    let mut reader = BufReader::new(file);
264    let items = read_all(&mut reader)?;
265
266    for item in items {
267        match item {
268            Item::PKCS8Key(bytes) => {
269                println!("Found PKCS8 key");
270                return Ok(PrivateKeyDer::from(PrivatePkcs8KeyDer::from(bytes)));
271            }
272            Item::RSAKey(bytes) => {
273                println!("Found RSA key");
274                return Ok(PrivateKeyDer::from(PrivatePkcs1KeyDer::from(bytes)));
275            }
276            Item::ECKey(bytes) => {
277                println!("Found EC key");
278                return Ok(PrivateKeyDer::from(PrivateSec1KeyDer::from(bytes)));
279            }
280            _ => println!("Skipping non-key item"),
281        }
282    }
283
284    Err(anyhow::anyhow!("no private key found"))
285}
286
287pub(crate) fn parse_ip_port(url: &str) -> (String, u16) {
288    let clean = url.trim_end_matches('/');
289    let without_scheme = clean.split("://").nth(1).unwrap_or(clean);
290    let mut parts = without_scheme.split(':');
291    let ip = parts.next().unwrap_or("127.0.0.1").to_string();
292    let port = parts.next().and_then(|p| p.parse().ok()).unwrap_or(443);
293    (ip, port)
294}
holger_ron/exposed/http2_backend.rs

holger_ron/exposed/
http2_backend.rs
