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