forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
10.0 KiB
1//! Full boot sequence: config, store, git checks, gitconfig and hooks, routers, SSH task.
2//!
3//! Three servers run until one of them stops or a shutdown signal arrives:
4//! - public HTTP on `config.listen` (UI and smart-HTTP),
5//! - raw user-content origin on `config.user_content_listen` (raw.rs, explicit routes only),
6//! - internal hook API on the unix socket `data_dir/internal.sock`,
7//! - SSH on `config.ssh_listen` (forge_ssh, which has no shutdown hook, so it is aborted).
8
9use std::os::unix::fs::PermissionsExt;
10use std::path::PathBuf;
11use std::sync::Arc;
12use std::time::Duration;
13
14use forge_core::{Config, Store, secret};
15use forge_git::GitEnv;
16use forge_ssh::SshConfig;
17use tokio::net::{TcpListener, UnixListener};
18use tokio::signal::unix::{SignalKind, signal};
19use tokio::sync::Notify;
20use tokio::sync::watch;
21use topcoat::cookie::RouterBuilderCookieExt;
22use topcoat::router::tower::TowerLayer;
23use topcoat::router::{Compression, Router, RouterBuilderDiscoverExt, RouterService};
24
25use crate::internal;
26use crate::raw;
27use crate::ssh_backend::SshBackend;
28use crate::state::AppState;
29
30/// Minimum git version forge relies on (receive-pack and stateless-rpc behaviour).
31pub(crate) const MIN_GIT: (u32, u32, u32) = (2, 38, 0);
32
33/// Boots every component and serves until shutdown.
34pub async fn serve(config: Config) -> forge_core::Result<()> {
35 let listen = config.listen;
36 let user_content_listen = config.user_content_listen;
37 let ssh_listen = config.ssh_listen;
38
39 std::fs::create_dir_all(&config.data_dir)?;
40 // The data dir holds the instance key and the internal socket. Keep it private to this user.
41 std::fs::set_permissions(&config.data_dir, std::fs::Permissions::from_mode(0o700))?;
42 let data_dir = config.data_dir.canonicalize()?;
43
44 let key = Arc::new(secret::load_or_create_key(&data_dir.join("forge.key"))?);
45 let store = Store::open(&config.database).await?;
46 store.set_secret_key(*key);
47
48 let git = Arc::new(GitEnv {
49 git_bin: find_git(),
50 data_dir: data_dir.clone(),
51 });
52 forge_git::version::require_min_version(&git, MIN_GIT).await?;
53 let version = forge_git::version::git_version(&git).await?;
54 forge_git::setup::write_managed_gitconfig(&data_dir)?;
55 let forge_bin = std::env::current_exe()?.canonicalize()?;
56 forge_git::setup::write_hook_shims(&data_dir, &forge_bin)?;
57
58 let internal_token: Arc<str> = Arc::from(random_hex());
59 let admin_token: Arc<str> = Arc::from(random_hex());
60 crate::admin_client::write_admin_token(&data_dir, &admin_token)?;
61 let internal_socket = data_dir.join("internal.sock");
62 // sun_path holds at most 108 bytes including the NUL terminator.
63 if internal_socket.as_os_str().len() >= 108 {
64 return Err(forge_core::Error::Config(format!(
65 "internal socket path is too long for a unix socket (max 107 bytes): {}; choose a shorter data_dir",
66 internal_socket.display()
67 )));
68 }
69 match std::fs::remove_file(&internal_socket) {
70 Ok(()) => {}
71 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
72 Err(e) => return Err(e.into()),
73 }
74 let unix = UnixListener::bind(&internal_socket)?;
75 std::fs::set_permissions(&internal_socket, std::fs::Permissions::from_mode(0o600))?;
76
77 let mailer: Arc<dyn crate::mail::Mailer> = match config.mail.transport {
78 forge_core::config::MailTransport::Smtp => {
79 Arc::new(crate::mail::SmtpMailer::new(store.clone()))
80 }
81 forge_core::config::MailTransport::Log => {
82 Arc::new(crate::mail::LogMailer::new(data_dir.join("mail-log")))
83 }
84 };
85 let state = AppState {
86 config: Arc::new(config),
87 store,
88 git,
89 key,
90 internal_token,
91 admin_token,
92 jobs: Arc::new(crate::backup::Jobs::default()),
93 internal_socket: internal_socket.clone(),
94 throttle: Arc::new(crate::throttle::Throttle::default()),
95 webhook_waker: Arc::new(Notify::new()),
96 mirror_waker: Arc::new(Notify::new()),
97 blobs: crate::blobs::BlobStore::open(&data_dir.join("blobs"))?,
98 replication: Arc::new(crate::replication::ReplicationQueue::default()),
99 ci_waker: Arc::new(Notify::new()),
100 mailer,
101 mail_waker: Arc::new(Notify::new()),
102 };
103 if let Err(e) = crate::signatures::rebuild_allowed_signers(&state).await {
104 tracing::error!(error = %e, "building the SSH allowed-signers file failed");
105 }
106 if let Err(e) = crate::signatures::rebuild_gpg_keyring(&state).await {
107 tracing::error!(error = %e, "rebuilding the GPG keyring failed");
108 }
109 crate::webhooks::worker::spawn(state.clone());
110 crate::mirrors::worker::spawn(state.clone());
111 crate::advisories::worker::spawn(state.clone());
112 crate::mail::worker::spawn(state.clone());
113 if state.config.backup.enabled() {
114 crate::backup::scheduler::spawn(state.clone());
115 }
116 if state.config.replication.enabled() {
117 crate::replication::worker::spawn(state.clone());
118 }
119 crate::ci::reaper::spawn(state.clone());
120
121 // Topcoat's builder-wide compression is off because it cannot filter by content type or path.
122 // `ResponseCompression` compresses text responses only and skips git, LFS, raw and archive paths.
123 let public: Router = Router::builder()
124 .discover()
125 .app_context(state.clone())
126 .cookies()
127 .layer(crate::pages::settings_profile::avatar_body_limit())
128 .layer(crate::pages::repo_ctx::RepoMovedLayer)
129 .layer(TowerLayer::new(crate::compression::ResponseCompression))
130 .compression(Compression::off())
131 .build();
132 let internal_router = internal::router(state.clone());
133 let raw_router = raw::router(state.clone());
134
135 let tcp = TcpListener::bind(listen).await?;
136 let raw_tcp = TcpListener::bind(user_content_listen).await.map_err(|e| {
137 forge_core::Error::Config(format!(
138 "cannot bind the user-content listener {user_content_listen}: {e}"
139 ))
140 })?;
141
142 tracing::info!(
143 listen = %listen,
144 user_content = %user_content_listen,
145 ssh = %ssh_listen,
146 data_dir = %data_dir.display(),
147 git = %format_version(version),
148 "forge starting"
149 );
150
151 let (stop_tx, stop_rx) = watch::channel(false);
152
153 let ssh_config = SshConfig {
154 listen: ssh_listen,
155 host_key_path: data_dir.join("ssh_host_ed25519_key"),
156 auth_timeout: Duration::from_secs(30),
157 inactivity_timeout: Duration::from_secs(10 * 60),
158 };
159 let ssh_backend = Arc::new(SshBackend {
160 state: state.clone(),
161 });
162 let ssh_stop = stop_tx.clone();
163 let mut ssh_task = tokio::spawn(async move {
164 let result = forge_ssh::run(ssh_config, ssh_backend).await;
165 // An SSH failure brings the whole process down.
166 ssh_stop.send_replace(true);
167 result
168 });
169
170 let signal_stop = stop_tx.clone();
171 tokio::spawn(async move {
172 shutdown_signal().await;
173 tracing::info!("shutdown signal received");
174 signal_stop.send_replace(true);
175 });
176
177 let http_stop = stop_tx.clone();
178 let http_rx = stop_rx.clone();
179 let http = async move {
180 let result =
181 topcoat::serve_until(tcp, RouterService::new(public), wait_stop(http_rx)).await;
182 http_stop.send_replace(true);
183 result
184 };
185
186 let internal_stop = stop_tx.clone();
187 let internal_rx = stop_rx.clone();
188 let internal_server = async move {
189 let result = topcoat::serve_until(
190 unix,
191 RouterService::new(internal_router),
192 wait_stop(internal_rx),
193 )
194 .await;
195 internal_stop.send_replace(true);
196 result
197 };
198
199 let raw_stop = stop_tx.clone();
200 let raw_rx = stop_rx.clone();
201 let raw_server = async move {
202 let result =
203 topcoat::serve_until(raw_tcp, RouterService::new(raw_router), wait_stop(raw_rx)).await;
204 raw_stop.send_replace(true);
205 result
206 };
207
208 let (http_result, internal_result, raw_result) =
209 tokio::join!(http, internal_server, raw_server);
210
211 // forge_ssh::run has no shutdown hook. If it is still running it is cancelled here.
212 ssh_task.abort();
213 let ssh_result: std::io::Result<()> = match (&mut ssh_task).await {
214 Ok(result) => result.map_err(|e| std::io::Error::other(e.to_string())),
215 Err(e) if e.is_cancelled() => Ok(()),
216 Err(e) => Err(std::io::Error::other(e.to_string())),
217 };
218
219 let _ = std::fs::remove_file(&internal_socket);
220 tracing::info!("forge stopped");
221
222 http_result?;
223 internal_result?;
224 raw_result?;
225 ssh_result?;
226 Ok(())
227}
228
229/// Resolves when the shared stop flag is set.
230async fn wait_stop(mut rx: watch::Receiver<bool>) {
231 let _ = rx.wait_for(|stopped| *stopped).await;
232}
233
234/// Resolves on ctrl_c or SIGTERM.
235async fn shutdown_signal() {
236 let ctrl_c = async {
237 if let Err(e) = tokio::signal::ctrl_c().await {
238 tracing::warn!(error = %e, "ctrl_c handler unavailable");
239 std::future::pending::<()>().await;
240 }
241 };
242 let terminate = async {
243 match signal(SignalKind::terminate()) {
244 Ok(mut term) => {
245 term.recv().await;
246 }
247 Err(e) => {
248 tracing::warn!(error = %e, "SIGTERM handler unavailable");
249 std::future::pending::<()>().await;
250 }
251 }
252 };
253 tokio::select! {
254 () = ctrl_c => {}
255 () = terminate => {}
256 }
257}
258
259/// Finds the git binary on PATH, falling back to /usr/bin/git.
260pub(crate) fn find_git() -> PathBuf {
261 std::env::var_os("PATH")
262 .and_then(|paths| {
263 std::env::split_paths(&paths)
264 .map(|dir| dir.join("git"))
265 .find(|candidate| candidate.is_file())
266 })
267 .and_then(|candidate| candidate.canonicalize().ok())
268 .unwrap_or_else(|| PathBuf::from("/usr/bin/git"))
269}
270
271/// 32 random bytes as lowercase hex. Used as the per-boot internal API token.
272fn random_hex() -> String {
273 let mut buf = [0u8; 32];
274 rand::fill(&mut buf);
275 hex::encode(buf)
276}
277
278fn format_version((major, minor, patch): (u32, u32, u32)) -> String {
279 format!("{major}.{minor}.{patch}")
280}