alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod exec;
2mod roster;
3mod server;
4
5use std::collections::BTreeSet;
6use std::net::SocketAddr;
7use std::path::Path;
8use std::sync::Arc;
9use std::time::Duration;
10
11use knot_atproto::Atproto;
12use knot_events::EventLog;
13use knot_git::Layout;
14use knot_index::Index;
15use knot_maintenance::MaintenanceHandle;
16use knot_pack::{MaxWireBytes, PackLimits};
17use knot_postreceive::LanguagesPushBudget;
18use knot_runtime::{Clock, Entropy, HttpTransport, OsEntropy};
19use knot_types::{AccountDid, ActorId, AdmissionPolicy, AppviewEndpoint, CiLogsAddr, KnotHostname};
20use russh::keys::ssh_key::rand_core;
21use russh::keys::{Algorithm, PrivateKey, ssh_key};
22use russh::server::{Config, Server as _};
23use russh::{MethodKind, MethodSet};
24use tokio::sync::Semaphore;
25use tokio_util::sync::CancellationToken;
26use tokio_util::task::TaskTracker;
27
28use knot_resource::{LimitConfig, PerPeerInflight, PreAuthLimiter, Slots};
29use roster::KeyRoster;
30use server::KnotSshServer;
31
32const MAX_INFLIGHT_PER_PEER: usize = 4;
33const INACTIVITY_TIMEOUT: Duration = Duration::from_secs(120);
34const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
35const AUTH_REJECTION_TIME: Duration = Duration::from_millis(250);
36const DRAIN_GRACE: Duration = Duration::from_secs(30);
37const ACCEPT_BACKOFF: Duration = Duration::from_millis(250);
38
39#[derive(Debug, thiserror::Error)]
40pub enum SshError {
41 #[error("ssh host key {path}: {message}")]
42 HostKey { path: String, message: String },
43 #[error("ssh server bind or serve: {0}")]
44 Serve(#[from] std::io::Error),
45}
46
47pub struct SshState<H, C> {
48 layout: Layout,
49 index: Arc<Index>,
50 atproto: Arc<Atproto<H, C>>,
51 knot_actor: ActorId,
52 events: Arc<EventLog<C>>,
53 hostname: KnotHostname,
54 appview: AppviewEndpoint,
55 admins: BTreeSet<AccountDid>,
56 admission: AdmissionPolicy,
57 limits: PackLimits,
58 max_pack_bytes: MaxWireBytes,
59 languages_push_budget: LanguagesPushBudget,
60 ci_logs: Option<CiLogsAddr>,
61 slots: Slots,
62 peer_slots: Arc<PreAuthLimiter>,
63 roster: Arc<KeyRoster>,
64 maintenance: MaintenanceHandle,
65 lfs: Option<LfsRuntime>,
66 catalog: Arc<knot_messages::Catalog>,
67}
68
69#[derive(Clone)]
70pub(crate) struct LfsRuntime {
71 pub(crate) handle: knot_lfs::LfsHandle,
72 pub(crate) slots: Arc<Semaphore>,
73 pub(crate) peer_slots: Arc<PreAuthLimiter>,
74}
75
76impl<H: HttpTransport, C: Clock> SshState<H, C> {
77 #[allow(clippy::too_many_arguments)]
78 pub fn new(
79 layout: Layout,
80 index: Arc<Index>,
81 atproto: Arc<Atproto<H, C>>,
82 knot_actor: ActorId,
83 events: Arc<EventLog<C>>,
84 hostname: KnotHostname,
85 appview: AppviewEndpoint,
86 admins: BTreeSet<AccountDid>,
87 admission: AdmissionPolicy,
88 max_pack_bytes: MaxWireBytes,
89 languages_push_budget: LanguagesPushBudget,
90 ci_logs: Option<CiLogsAddr>,
91 ) -> Self {
92 Self {
93 layout,
94 index,
95 atproto,
96 knot_actor,
97 events,
98 hostname,
99 appview,
100 admins,
101 admission,
102 limits: PackLimits::default(),
103 max_pack_bytes,
104 languages_push_budget,
105 ci_logs,
106 slots: Slots::for_machine(),
107 peer_slots: Arc::new(PreAuthLimiter::with_config(LimitConfig::per_peer_only(
108 PerPeerInflight::new(MAX_INFLIGHT_PER_PEER),
109 ))),
110 roster: Arc::new(KeyRoster::new()),
111 maintenance: MaintenanceHandle::disabled(),
112 lfs: None,
113 catalog: Arc::new(knot_messages::Catalog::defaults()),
114 }
115 }
116
117 pub fn with_catalog(mut self, catalog: Arc<knot_messages::Catalog>) -> Self {
118 self.catalog = catalog;
119 self
120 }
121
122 pub fn with_slots(mut self, slots: Slots) -> Self {
123 self.slots = slots;
124 self
125 }
126
127 pub fn with_maintenance(mut self, maintenance: MaintenanceHandle) -> Self {
128 self.maintenance = maintenance;
129 self
130 }
131
132 pub fn with_lfs(mut self, handle: knot_lfs::LfsHandle, max_transfers: usize) -> Self {
133 self.lfs = Some(LfsRuntime {
134 handle,
135 slots: Arc::new(Semaphore::new(max_transfers)),
136 peer_slots: Arc::new(PreAuthLimiter::with_config(LimitConfig::per_peer_only(
137 PerPeerInflight::new(max_transfers),
138 ))),
139 });
140 self
141 }
142
143 pub fn with_limits(mut self, limits: PackLimits) -> Self {
144 self.limits = limits;
145 self
146 }
147}
148
149fn server_config(host_key: PrivateKey) -> Arc<Config> {
150 Arc::new(Config {
151 keys: vec![host_key],
152 methods: MethodSet::from(&[MethodKind::PublicKey][..]),
153 inactivity_timeout: Some(INACTIVITY_TIMEOUT),
154 keepalive_interval: Some(KEEPALIVE_INTERVAL),
155 auth_rejection_time: AUTH_REJECTION_TIME,
156 ..Config::default()
157 })
158}
159
160pub async fn serve<H: HttpTransport, C: Clock>(
161 addr: SocketAddr,
162 host_key: PrivateKey,
163 state: Arc<SshState<H, C>>,
164 shutdown: CancellationToken,
165) -> Result<(), SshError> {
166 let listener = tokio::net::TcpListener::bind(addr).await?;
167 serve_drained(listener, host_key, state, shutdown).await
168}
169
170pub async fn serve_on_socket<H: HttpTransport, C: Clock>(
171 listener: tokio::net::TcpListener,
172 host_key: PrivateKey,
173 state: Arc<SshState<H, C>>,
174) -> Result<(), SshError> {
175 serve_drained(listener, host_key, state, CancellationToken::new()).await
176}
177
178#[doc(hidden)]
179pub async fn serve_drained<H: HttpTransport, C: Clock>(
180 listener: tokio::net::TcpListener,
181 host_key: PrivateKey,
182 state: Arc<SshState<H, C>>,
183 shutdown: CancellationToken,
184) -> Result<(), SshError> {
185 let config = server_config(host_key);
186 let tracker = TaskTracker::new();
187 state.roster.prime(&state.index, &state.atproto);
188 let mut server = KnotSshServer {
189 state,
190 tracker: tracker.clone(),
191 };
192 loop {
193 let accepted = tokio::select! {
194 () = shutdown.cancelled() => break,
195 accepted = listener.accept() => accepted,
196 };
197 let (stream, peer) = match accepted {
198 Ok(pair) => pair,
199 Err(error) if is_connection_error(&error) => continue,
200 Err(error) => {
201 tracing::warn!("ssh accept failed, backing off: {error}");
202 tokio::select! {
203 () = shutdown.cancelled() => break,
204 () = tokio::time::sleep(ACCEPT_BACKOFF) => {}
205 }
206 continue;
207 }
208 };
209 let handler = server.new_client(Some(peer));
210 let config = Arc::clone(&config);
211 tracker.spawn(async move {
212 if let Ok(session) = russh::server::run_stream(config, stream, handler).await {
213 let _ = session.await;
214 }
215 });
216 }
217 tracker.close();
218 let _ = tokio::time::timeout(DRAIN_GRACE, tracker.wait()).await;
219 Ok(())
220}
221
222fn is_connection_error(error: &std::io::Error) -> bool {
223 matches!(
224 error.kind(),
225 std::io::ErrorKind::ConnectionRefused
226 | std::io::ErrorKind::ConnectionAborted
227 | std::io::ErrorKind::ConnectionReset
228 )
229}
230
231pub fn load_or_create_host_key(path: &Path) -> Result<PrivateKey, SshError> {
232 let report = |message: String| SshError::HostKey {
233 path: path.display().to_string(),
234 message,
235 };
236 let load = || {
237 ensure_secure_perms(path).map_err(&report)?;
238 russh::keys::load_secret_key(path, None).map_err(|error| report(error.to_string()))
239 };
240 if path.exists() {
241 return load();
242 }
243 let key = PrivateKey::random(&mut EntropyRng, Algorithm::Ed25519)
244 .map_err(|error| report(error.to_string()))?;
245 match persist_host_key(path, &key)? {
246 Claim::Won => Ok(key),
247 Claim::Lost => load(),
248 }
249}
250
251enum Claim {
252 Won,
253 Lost,
254}
255
256fn persist_host_key(path: &Path, key: &PrivateKey) -> Result<Claim, SshError> {
257 let report = |message: String| SshError::HostKey {
258 path: path.display().to_string(),
259 message,
260 };
261 let pem = key
262 .to_openssh(ssh_key::LineEnding::LF)
263 .map_err(|error| report(error.to_string()))?;
264 if let Some(parent) = path.parent() {
265 std::fs::create_dir_all(parent).map_err(|error| report(error.to_string()))?;
266 }
267 let temp = unique_temp(path);
268 write_secret(&temp, pem.as_bytes()).map_err(|error| report(error.to_string()))?;
269 // `hard_link` instead of `rename` so that in case two knots
270 // are booting at the same time they don't get borked
271 // if one clobber's the other's key. here the loser reloads
272 // winner's key.
273 let claim = match std::fs::hard_link(&temp, path) {
274 Ok(()) => Claim::Won,
275 Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Claim::Lost,
276 Err(error) => {
277 let _ = std::fs::remove_file(&temp);
278 return Err(report(error.to_string()));
279 }
280 };
281 let _ = std::fs::remove_file(&temp);
282 if let Some(parent) = path.parent()
283 && let Ok(dir) = std::fs::File::open(parent)
284 {
285 let _ = dir.sync_all();
286 }
287 Ok(claim)
288}
289
290#[cfg(unix)]
291fn ensure_secure_perms(path: &Path) -> Result<(), String> {
292 use std::os::unix::fs::MetadataExt;
293 let mode = std::fs::metadata(path)
294 .map_err(|error| error.to_string())?
295 .mode();
296 if mode & 0o077 != 0 {
297 return Err(format!(
298 "private host key is group or other accessible at mode {:o}, run chmod 600 on it",
299 mode & 0o777
300 ));
301 }
302 Ok(())
303}
304
305#[cfg(not(unix))]
306fn ensure_secure_perms(_path: &Path) -> Result<(), String> {
307 Ok(())
308}
309
310fn unique_temp(path: &Path) -> std::path::PathBuf {
311 use std::sync::atomic::{AtomicU64, Ordering};
312 static COUNTER: AtomicU64 = AtomicU64::new(0);
313 let nonce = COUNTER.fetch_add(1, Ordering::Relaxed);
314 let stem = path
315 .file_name()
316 .and_then(|name| name.to_str())
317 .unwrap_or("ssh_host_key");
318 path.with_file_name(format!(".{stem}.{}.{nonce}.tmp", std::process::id()))
319}
320
321fn write_secret(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
322 use std::io::Write;
323 let mut options = std::fs::OpenOptions::new();
324 options.write(true).create_new(true);
325 #[cfg(unix)]
326 {
327 use std::os::unix::fs::OpenOptionsExt;
328 options.mode(0o600);
329 }
330 let mut file = options.open(path)?;
331 file.write_all(bytes)?;
332 file.sync_all()
333}
334
335struct EntropyRng;
336
337impl rand_core::TryRng for EntropyRng {
338 type Error = std::convert::Infallible;
339
340 fn try_next_u32(&mut self) -> Result<u32, Self::Error> {
341 let mut bytes = [0u8; 4];
342 OsEntropy.fill(&mut bytes);
343 Ok(u32::from_le_bytes(bytes))
344 }
345
346 fn try_next_u64(&mut self) -> Result<u64, Self::Error> {
347 let mut bytes = [0u8; 8];
348 OsEntropy.fill(&mut bytes);
349 Ok(u64::from_le_bytes(bytes))
350 }
351
352 fn try_fill_bytes(&mut self, dst: &mut [u8]) -> Result<(), Self::Error> {
353 OsEntropy.fill(dst);
354 Ok(())
355 }
356}
357
358impl rand_core::TryCryptoRng for EntropyRng {}