| 1 | |
| 2 | |
| 3 | |
| 4 | |
| 5 | |
| 6 | |
| 7 | |
| 8 | |
| 9 | use std::io::Write; |
| 10 | use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt, PermissionsExt}; |
| 11 | use std::path::{Path, PathBuf}; |
| 12 | |
| 13 | use forge_core::{Config, Error, Result}; |
| 14 | use serde::de::DeserializeOwned; |
| 15 | use serde_json::Value; |
| 16 | use tokio::io::{AsyncReadExt, AsyncWriteExt}; |
| 17 | use tokio::net::UnixStream; |
| 18 | |
| 19 | |
| 20 | pub const ADMIN_TOKEN_HEADER: &str = "x-forge-admin-token"; |
| 21 | const RUN_DIR: &str = "run"; |
| 22 | const TOKEN_FILE: &str = "admin.token"; |
| 23 | const SOCKET_FILE: &str = "internal.sock"; |
| 24 | |
| 25 | |
| 26 | |
| 27 | pub fn write_admin_token(data_dir: &Path, token: &str) -> Result<()> { |
| 28 | let run = data_dir.join(RUN_DIR); |
| 29 | std::fs::DirBuilder::new() |
| 30 | .recursive(true) |
| 31 | .mode(0o700) |
| 32 | .create(&run)?; |
| 33 | std::fs::set_permissions(&run, std::fs::Permissions::from_mode(0o700))?; |
| 34 | let path = run.join(TOKEN_FILE); |
| 35 | match std::fs::remove_file(&path) { |
| 36 | Ok(()) => {} |
| 37 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} |
| 38 | Err(e) => return Err(e.into()), |
| 39 | } |
| 40 | let mut file = std::fs::OpenOptions::new() |
| 41 | .write(true) |
| 42 | .create_new(true) |
| 43 | .mode(0o600) |
| 44 | .open(&path)?; |
| 45 | file.write_all(token.as_bytes())?; |
| 46 | file.sync_all()?; |
| 47 | Ok(()) |
| 48 | } |
| 49 | |
| 50 | |
| 51 | pub fn admin_token_path(data_dir: &Path) -> PathBuf { |
| 52 | data_dir.join(RUN_DIR).join(TOKEN_FILE) |
| 53 | } |
| 54 | |
| 55 | |
| 56 | pub struct AdminClient { |
| 57 | socket: PathBuf, |
| 58 | token: String, |
| 59 | } |
| 60 | |
| 61 | |
| 62 | |
| 63 | |
| 64 | pub async fn running_server(config: &Config) -> Result<Option<AdminClient>> { |
| 65 | let socket = config.data_dir.join(SOCKET_FILE); |
| 66 | if !socket.exists() { |
| 67 | return Ok(None); |
| 68 | } |
| 69 | match UnixStream::connect(&socket).await { |
| 70 | Ok(stream) => drop(stream), |
| 71 | Err(e) |
| 72 | if matches!( |
| 73 | e.kind(), |
| 74 | std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound |
| 75 | ) => |
| 76 | { |
| 77 | return Ok(None); |
| 78 | } |
| 79 | Err(e) => return Err(e.into()), |
| 80 | } |
| 81 | let path = admin_token_path(&config.data_dir); |
| 82 | let token = std::fs::read_to_string(&path).map_err(|e| { |
| 83 | Error::Config(format!( |
| 84 | "a server is running, but its admin token at {} cannot be read: {e}", |
| 85 | path.display() |
| 86 | )) |
| 87 | })?; |
| 88 | Ok(Some(AdminClient { |
| 89 | socket, |
| 90 | token: token.trim().to_owned(), |
| 91 | })) |
| 92 | } |
| 93 | |
| 94 | #[derive(serde::Deserialize)] |
| 95 | struct ErrorBody { |
| 96 | error: String, |
| 97 | } |
| 98 | |
| 99 | impl AdminClient { |
| 100 | |
| 101 | |
| 102 | pub async fn post<T: DeserializeOwned>(&self, path: &str, body: &Value) -> Result<T> { |
| 103 | let payload = body.to_string(); |
| 104 | let mut stream = UnixStream::connect(&self.socket).await?; |
| 105 | let head = format!( |
| 106 | "POST {path} HTTP/1.1\r\n\ |
| 107 | Host: forge-internal\r\n\ |
| 108 | Content-Type: application/json\r\n\ |
| 109 | Content-Length: {}\r\n\ |
| 110 | {ADMIN_TOKEN_HEADER}: {}\r\n\ |
| 111 | Connection: close\r\n\r\n", |
| 112 | payload.len(), |
| 113 | self.token |
| 114 | ); |
| 115 | stream.write_all(head.as_bytes()).await?; |
| 116 | stream.write_all(payload.as_bytes()).await?; |
| 117 | stream.flush().await?; |
| 118 | let mut raw = Vec::new(); |
| 119 | stream.read_to_end(&mut raw).await?; |
| 120 | let (status, bytes) = crate::hook_cmd::parse_response(&raw)?; |
| 121 | if status == 200 { |
| 122 | return serde_json::from_slice(&bytes) |
| 123 | .map_err(|e| Error::Invalid(format!("unexpected answer from the server: {e}"))); |
| 124 | } |
| 125 | let message = serde_json::from_slice::<ErrorBody>(&bytes) |
| 126 | .map(|b| b.error) |
| 127 | .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).trim().to_owned()); |
| 128 | Err(match status { |
| 129 | 403 => Error::Forbidden(message), |
| 130 | 409 => Error::Conflict(message), |
| 131 | 400 => Error::Invalid(message), |
| 132 | _ => Error::Git(format!("server answered {status}: {message}")), |
| 133 | }) |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | #[cfg(test)] |
| 138 | mod tests { |
| 139 | use super::*; |
| 140 | use crate::backup::FsckSummary; |
| 141 | use crate::test_support::{ADMIN_TOKEN, test_config, test_state}; |
| 142 | use topcoat::router::RouterService; |
| 143 | |
| 144 | #[test] |
| 145 | fn token_file_is_private_and_replaced_on_each_boot() { |
| 146 | let dir = tempfile::tempdir().expect("tempdir"); |
| 147 | write_admin_token(dir.path(), "first").expect("write"); |
| 148 | let path = admin_token_path(dir.path()); |
| 149 | assert_eq!(std::fs::read_to_string(&path).expect("read"), "first"); |
| 150 | assert_eq!( |
| 151 | std::fs::metadata(&path).expect("meta").permissions().mode() & 0o777, |
| 152 | 0o600 |
| 153 | ); |
| 154 | assert_eq!( |
| 155 | std::fs::metadata(path.parent().expect("run dir")) |
| 156 | .expect("meta") |
| 157 | .permissions() |
| 158 | .mode() |
| 159 | & 0o777, |
| 160 | 0o700 |
| 161 | ); |
| 162 | write_admin_token(dir.path(), "second").expect("rewrite"); |
| 163 | assert_eq!(std::fs::read_to_string(&path).expect("read"), "second"); |
| 164 | } |
| 165 | |
| 166 | #[tokio::test] |
| 167 | async fn no_socket_means_no_server() { |
| 168 | let dir = tempfile::tempdir().expect("tempdir"); |
| 169 | let config = crate::test_support::test_config(dir.path()); |
| 170 | assert!(running_server(&config).await.expect("probe").is_none()); |
| 171 | } |
| 172 | |
| 173 | #[tokio::test] |
| 174 | async fn stale_socket_means_no_server() { |
| 175 | let dir = tempfile::tempdir().expect("tempdir"); |
| 176 | let config = crate::test_support::test_config(dir.path()); |
| 177 | |
| 178 | let listener = |
| 179 | std::os::unix::net::UnixListener::bind(dir.path().join(SOCKET_FILE)).expect("bind"); |
| 180 | drop(listener); |
| 181 | assert!(running_server(&config).await.expect("probe").is_none()); |
| 182 | } |
| 183 | |
| 184 | |
| 185 | |
| 186 | #[tokio::test] |
| 187 | async fn client_talks_to_the_server_over_the_socket() { |
| 188 | let dir = tempfile::tempdir().expect("tempdir"); |
| 189 | let data_dir = dir.path().join("data"); |
| 190 | std::fs::create_dir_all(&data_dir).expect("data dir"); |
| 191 | let state = test_state(&data_dir).await; |
| 192 | write_admin_token(&data_dir, ADMIN_TOKEN).expect("token"); |
| 193 | let listener = tokio::net::UnixListener::bind(data_dir.join(SOCKET_FILE)).expect("bind"); |
| 194 | let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); |
| 195 | let server = tokio::spawn(topcoat::serve_until( |
| 196 | listener, |
| 197 | RouterService::new(crate::internal::router(state)), |
| 198 | async move { |
| 199 | let _ = stop_rx.await; |
| 200 | }, |
| 201 | )); |
| 202 | |
| 203 | let config = test_config(&data_dir); |
| 204 | let client = running_server(&config) |
| 205 | .await |
| 206 | .expect("probe") |
| 207 | .expect("the server is up"); |
| 208 | let summary: FsckSummary = client |
| 209 | .post( |
| 210 | "/internal/admin/fsck", |
| 211 | &serde_json::json!({ "full": false }), |
| 212 | ) |
| 213 | .await |
| 214 | .expect("fsck through the socket"); |
| 215 | assert!(summary.is_ok()); |
| 216 | assert_eq!(summary.repos, 0); |
| 217 | |
| 218 | |
| 219 | let wrong = AdminClient { |
| 220 | socket: data_dir.join(SOCKET_FILE), |
| 221 | token: "not-the-token".into(), |
| 222 | }; |
| 223 | let refused: Result<FsckSummary> = wrong |
| 224 | .post("/internal/admin/fsck", &serde_json::json!({})) |
| 225 | .await; |
| 226 | assert!(matches!(refused, Err(Error::Forbidden(_))), "{refused:?}"); |
| 227 | |
| 228 | let _ = stop_tx.send(()); |
| 229 | let _ = server.await; |
| 230 | } |
| 231 | } |