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