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