dvs/backends/
server.rs

1use anyhow::bail;
2use fs_err as fs;
3use serde::{Deserialize, Serialize};
4use std::io::Read;
5use std::path::PathBuf;
6use std::sync::OnceLock;
7use tempfile::NamedTempFile;
8use ureq::Body;
9use ureq::http::Response;
10use url::Url;
11
12use crate::audit::AuditEntry;
13use crate::auth::get_token;
14use crate::error::{Error, ErrorCode};
15use crate::utils::http_agent;
16use crate::{Compression, Hashes, ProjectPath, RetrieveRequest, StoreRequest, StoreResult};
17
18const NOT_ALLOWED: &str =
19    "You are not allowed to access this project, make sure you are logged in with the right user.";
20const LOGIN_EXPIRED: &str = "Your login is no longer valid, run `dvs login`";
21
22/// Join `relative` under the server URL, keeping any proxy path prefix:
23/// `Url::join` drops it for a leading-slash path or a base without a
24/// trailing slash.
25pub(crate) fn join_url(base: &Url, relative: &str) -> anyhow::Result<Url> {
26    debug_assert!(!relative.starts_with('/'), "{relative} must be relative");
27    if base.cannot_be_a_base() {
28        bail!("{base} cannot be used as a dvs server URL");
29    }
30    let mut base = base.clone();
31    if !base.path().ends_with('/') {
32        base.set_path(&format!("{}/", base.path()));
33    }
34    base.join(relative)
35        .map_err(|e| anyhow::anyhow!("{base} cannot be used as a dvs server URL: {e}"))
36}
37
38/// The shared code for an HTTP status, `None` for endpoint-specific ones.
39///
40/// dvs-server answers a missing project with 404 and a body starting with
41/// `unknown project`; any other 404 (or a 405) is a route this server does
42/// not have yet.
43pub(crate) fn http_status_code(status: u16, body: &str) -> Option<ErrorCode> {
44    match status {
45        401 => Some(ErrorCode::AuthExpired),
46        403 => Some(ErrorCode::AccessDenied),
47        404 if body.trim_start().starts_with("unknown project") => Some(ErrorCode::ProjectNotFound),
48        404 | 405 => Some(ErrorCode::ServerTooOld),
49        500.. => Some(ErrorCode::ServerUnavailable),
50        _ => None,
51    }
52}
53
54fn coded(code: ErrorCode, message: impl Into<String>) -> anyhow::Error {
55    Error::new(code, message).into()
56}
57
58pub(crate) fn unreachable(base_url: &Url, e: ureq::Error) -> anyhow::Error {
59    coded(
60        ErrorCode::ServerUnavailable,
61        format!("cannot reach the dvs server at {base_url}: {e}"),
62    )
63}
64
65pub(crate) fn invalid_response(base_url: &Url, e: impl std::fmt::Display) -> anyhow::Error {
66    coded(
67        ErrorCode::ServerUnavailable,
68        format!("Invalid response from the dvs server at {base_url}: {e}"),
69    )
70}
71
72/// The error for a status the endpoint does not handle itself.
73pub(crate) fn status_error(base_url: &Url, status: u16, body: &str, what: &str) -> anyhow::Error {
74    let Some(code) = http_status_code(status, body) else {
75        return anyhow::anyhow!("{what} failed ({status}): {body}");
76    };
77    let message = match code {
78        ErrorCode::AuthExpired => LOGIN_EXPIRED.to_string(),
79        ErrorCode::AccessDenied => NOT_ALLOWED.to_string(),
80        ErrorCode::ProjectNotFound => format!(
81            "the dvs server at {base_url} does not have this project ({})",
82            body.trim()
83        ),
84        ErrorCode::ServerTooOld => format!(
85            "the dvs server at {base_url} does not support this request ({status}); it is older than this dvs"
86        ),
87        _ => format!("the dvs server at {base_url} is not working ({status})"),
88    };
89    coded(code, message)
90}
91
92pub(crate) fn response_error(
93    base_url: &Url,
94    resp: &mut Response<Body>,
95    what: &str,
96) -> anyhow::Error {
97    let body = resp.body_mut().read_to_string().unwrap_or_default();
98    status_error(base_url, resp.status().as_u16(), &body, what)
99}
100
101/// Wraps a reader and reports bytes as they're read, driving the `on_bytes`
102/// progress callback. Used to stream an upload body while updating a progress bar.
103///
104/// It also remembers its reader's first error: once ureq or `io::copy` has
105/// failed, that is the only way to tell a local read error from a network one.
106struct CountingReader<'a, R> {
107    inner: R,
108    on_bytes: Option<&'a (dyn Fn(u64) + Send + Sync)>,
109    error: Option<std::io::Error>,
110}
111
112impl<'a, R> CountingReader<'a, R> {
113    fn new(inner: R, on_bytes: Option<&'a (dyn Fn(u64) + Send + Sync)>) -> Self {
114        Self {
115            inner,
116            on_bytes,
117            error: None,
118        }
119    }
120
121    fn take_error(&mut self) -> Option<std::io::Error> {
122        self.error.take()
123    }
124}
125
126impl<R: Read> Read for CountingReader<'_, R> {
127    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
128        let n = match self.inner.read(buf) {
129            Ok(n) => n,
130            Err(e) => {
131                if self.error.is_none() && e.kind() != std::io::ErrorKind::Interrupted {
132                    self.error = Some(std::io::Error::new(e.kind(), e.to_string()));
133                }
134                return Err(e);
135            }
136        };
137        if n > 0 {
138            if let Some(cb) = self.on_bytes {
139                cb(n as u64);
140            }
141        }
142        Ok(n)
143    }
144}
145
146#[derive(Serialize, Deserialize)]
147pub struct InitPayload {
148    pub name: String,
149    pub group: String,
150    pub compression: Compression,
151}
152
153#[derive(Serialize, Deserialize)]
154pub struct UploadResponse {
155    pub stored_size: u64,
156    pub compression: Compression,
157}
158
159#[derive(Serialize, Deserialize)]
160pub struct HistoryPayload {
161    #[serde(default)]
162    pub paths: Vec<ProjectPath>,
163}
164
165#[derive(Serialize, Deserialize)]
166pub struct UploadManifestEntry {
167    pub path: ProjectPath,
168    pub size: u64,
169    pub hash: String,
170}
171
172#[derive(Serialize, Deserialize)]
173pub struct UploadManifest {
174    pub operation_id: uuid::Uuid,
175    pub message: Option<String>,
176    pub entries: Vec<UploadManifestEntry>,
177    /// git current status at the time of `dvs add`. The file may or may not
178    /// be in this branch/commit.
179    /// This is more to know: what was the person working on when they pushed
180    #[serde(default)]
181    pub git_branch: Option<String>,
182    #[serde(default)]
183    pub git_commit: Option<String>,
184}
185
186#[derive(Serialize, Deserialize)]
187pub struct UploadManifestResponse {
188    /// List of hashes we don't know about -> they need to be uploaded
189    pub missing: Vec<String>,
190    pub compression: Compression,
191}
192
193#[derive(Serialize, Deserialize)]
194pub struct AccessUser {
195    pub username: String,
196    pub email: Option<String>,
197}
198
199#[derive(Serialize, Deserialize)]
200pub struct AccessResponse {
201    pub access: String,
202    pub user: AccessUser,
203    pub groups: Vec<String>,
204}
205
206#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
207pub struct ServerBackend {
208    pub name: String,
209    pub url: Url,
210    /// This is used for init but never after since people can edit groups in the
211    /// UI so that field would be dead either way
212    #[serde(skip)]
213    pub group: Option<String>,
214    #[serde(skip)]
215    access_token: OnceLock<Option<String>>,
216}
217
218impl ServerBackend {
219    pub fn new(name: String, group: Option<String>, url: Url) -> Self {
220        Self {
221            name,
222            group,
223            url,
224            access_token: OnceLock::new(),
225        }
226    }
227
228    fn init_url(&self) -> anyhow::Result<Url> {
229        join_url(&self.url, "api/init")
230    }
231
232    fn project_url(&self) -> anyhow::Result<Url> {
233        join_url(&self.url, &format!("api/project/{}", self.name))
234    }
235
236    fn blob_url(&self, hash: &str) -> anyhow::Result<Url> {
237        self.read_url(&format!("blobs/{hash}"))
238    }
239
240    fn manifest_url(&self) -> anyhow::Result<Url> {
241        self.read_url("manifest")
242    }
243
244    fn access_url(&self) -> anyhow::Result<Url> {
245        self.read_url("access")
246    }
247
248    fn history_url(&self) -> anyhow::Result<Url> {
249        self.read_url("history")
250    }
251
252    fn bearer(&self) -> anyhow::Result<String> {
253        let token = match self.access_token.get() {
254            Some(token) => token,
255            None => {
256                let read = get_token(&self.url)?;
257                self.access_token.get_or_init(|| read)
258            }
259        };
260        match token {
261            Some(token) => Ok(format!("Bearer {token}")),
262            None => Err(coded(
263                ErrorCode::AuthRequired,
264                "Could not find access token. Log in first",
265            )),
266        }
267    }
268
269    fn read_url(&self, endpoint: &str) -> anyhow::Result<Url> {
270        join_url(&self.url, &format!("api/{}/{}", self.name, endpoint))
271    }
272
273    pub fn init(&self, compression: Compression) -> anyhow::Result<bool> {
274        let Some(group) = self.group.clone() else {
275            bail!("A group is required to create a project on the server");
276        };
277        let url = self.init_url()?;
278        let payload = InitPayload {
279            name: self.name.clone(),
280            group,
281            compression,
282        };
283        let mut resp = http_agent()
284            .post(url.as_str())
285            .header("authorization", self.bearer()?)
286            .send_json(&payload)
287            .map_err(|e| unreachable(&self.url, e))?;
288        match resp.status().as_u16() {
289            201 => Ok(false), // created
290            200 => Ok(true),  // already existed (group+compression matched)
291            409 => {
292                bail!(
293                    "Server initialization failed: the project already exists with a different group or compression"
294                )
295            }
296            _ => Err(response_error(&self.url, &mut resp, "server init")),
297        }
298    }
299
300    pub fn attach(&self) -> anyhow::Result<()> {
301        let url = self.project_url()?;
302        let mut resp = http_agent()
303            .get(url.as_str())
304            .header("authorization", self.bearer()?)
305            .call()
306            .map_err(|e| unreachable(&self.url, e))?;
307
308        match resp.status().as_u16() {
309            200 => Ok(()),
310            _ => Err(response_error(
311                &self.url,
312                &mut resp,
313                "getting project information",
314            )),
315        }
316    }
317
318    /// What the current user can do on the project. `None` when they are not a member.
319    pub fn access(&self) -> anyhow::Result<Option<AccessResponse>> {
320        let mut resp = http_agent()
321            .get(self.access_url()?.as_str())
322            .header("authorization", self.bearer()?)
323            .call()
324            .map_err(|e| unreachable(&self.url, e))?;
325
326        match resp.status().as_u16() {
327            200 => Ok(Some(
328                resp.body_mut()
329                    .read_json()
330                    .map_err(|e| invalid_response(&self.url, e))?,
331            )),
332            403 => Ok(None),
333            _ => Err(response_error(&self.url, &mut resp, "access check")),
334        }
335    }
336
337    pub fn check_access(&self) -> anyhow::Result<()> {
338        if self.access()?.is_none() {
339            return Err(coded(ErrorCode::AccessDenied, NOT_ALLOWED));
340        }
341        Ok(())
342    }
343
344    /// Declares what an add is about to store. The server records every entry and
345    /// answers with the hashes it does not have, which are the only ones to `store`.
346    pub fn manifest(&self, manifest: &UploadManifest) -> anyhow::Result<UploadManifestResponse> {
347        let mut resp = http_agent()
348            .post(self.manifest_url()?.as_str())
349            .header("authorization", self.bearer()?)
350            .send_json(manifest)
351            .map_err(|e| unreachable(&self.url, e))?;
352
353        match resp.status().as_u16() {
354            200 => resp
355                .body_mut()
356                .read_json()
357                .map_err(|e| invalid_response(&self.url, e)),
358            _ => Err(response_error(&self.url, &mut resp, "manifest")),
359        }
360    }
361
362    pub fn store(&self, req: StoreRequest<'_>) -> anyhow::Result<StoreResult> {
363        let mut file = CountingReader::new(fs::File::open(req.source)?, req.on_bytes);
364
365        let mut url = self.blob_url(req.hashes.get_blake3())?;
366        url.query_pairs_mut()
367            .append_pair("operation_id", &req.operation_id.to_string());
368        let sent = http_agent()
369            .put(url.as_str())
370            .header("authorization", self.bearer()?)
371            .header("content-type", "application/octet-stream")
372            .send(ureq::SendBody::from_reader(&mut file));
373        let mut resp = match sent {
374            Ok(resp) => resp,
375            Err(e) => match file.take_error() {
376                Some(local) => return Err(local.into()),
377                None => return Err(unreachable(&self.url, e)),
378            },
379        };
380
381        match resp.status().as_u16() {
382            status @ (200 | 201) => {
383                let uploaded: UploadResponse = resp
384                    .body_mut()
385                    .read_json()
386                    .map_err(|e| invalid_response(&self.url, e))?;
387                Ok(StoreResult {
388                    stored_size: uploaded.stored_size,
389                    compression: uploaded.compression,
390                    created: status == 201,
391                })
392            }
393            _ => Err(response_error(&self.url, &mut resp, "upload")),
394        }
395    }
396
397    pub fn retrieve(&self, req: RetrieveRequest<'_>) -> anyhow::Result<bool> {
398        let mut url = self.blob_url(req.hashes.get_blake3())?;
399        url.query_pairs_mut()
400            .append_pair("path", req.path.as_str())
401            .append_pair("operation_id", &req.operation_id.to_string());
402
403        let mut resp = http_agent()
404            .get(url.as_str())
405            .header("authorization", self.bearer()?)
406            .call()
407            .map_err(|e| unreachable(&self.url, e))?;
408
409        match resp.status().as_u16() {
410            200 => {}
411            code => {
412                let body = resp.body_mut().read_to_string().unwrap_or_default();
413                // On a blob, 404 is "no such content" unless the project is missing
414                if code == 404 && http_status_code(code, &body) != Some(ErrorCode::ProjectNotFound)
415                {
416                    return Ok(false);
417                }
418                return Err(status_error(&self.url, code, &body, "download"));
419            }
420        }
421
422        let tmp = NamedTempFile::new()?.into_temp_path();
423        {
424            let mut reader = CountingReader::new(resp.body_mut().as_reader(), req.on_bytes);
425            let mut out = fs::File::create(&tmp)?;
426            self.download(&mut reader, &mut out)?;
427        }
428
429        if let Some(parent) = req.target.parent() {
430            fs::create_dir_all(parent)?;
431        }
432        req.compression.decompress(&tmp, req.target, None)?;
433        Ok(true)
434    }
435
436    fn download<R: Read>(
437        &self,
438        body: &mut CountingReader<'_, R>,
439        out: &mut impl std::io::Write,
440    ) -> anyhow::Result<u64> {
441        std::io::copy(body, out).map_err(|e| match body.take_error() {
442            Some(network) => coded(
443                ErrorCode::ServerUnavailable,
444                format!("the download from {} was cut off: {network}", self.url),
445            ),
446            None => e.into(),
447        })
448    }
449
450    pub fn exists(&self, hashes: &Hashes) -> anyhow::Result<bool> {
451        let url = self.blob_url(hashes.get_blake3())?;
452
453        match http_agent()
454            .head(url.as_str())
455            .header("authorization", self.bearer()?)
456            .call()
457            .map_err(|e| unreachable(&self.url, e))?
458            .status()
459            .as_u16()
460        {
461            200 => Ok(true),
462            404 => Ok(false),
463            code => Err(status_error(
464                &self.url,
465                code,
466                "",
467                "checking whether a file exists",
468            )),
469        }
470    }
471
472    pub fn remove(&self, _: &Hashes) -> anyhow::Result<()> {
473        // We don't rollback with the server
474        Ok(())
475    }
476
477    pub fn get_audit_entries(&self, files: &[PathBuf]) -> anyhow::Result<Vec<AuditEntry>> {
478        let url = self.history_url()?;
479
480        let mut paths = Vec::with_capacity(files.len());
481        for p in files {
482            paths.push(ProjectPath::from_path(p).map_err(|e| anyhow::anyhow!(e))?);
483        }
484
485        let payload = HistoryPayload { paths };
486
487        let mut resp = http_agent()
488            .post(url.as_str())
489            .header("authorization", self.bearer()?)
490            .send_json(&payload)
491            .map_err(|e| unreachable(&self.url, e))?;
492
493        match resp.status().as_u16() {
494            200 => resp
495                .body_mut()
496                .read_json::<Vec<AuditEntry>>()
497                .map_err(|e| invalid_response(&self.url, e)),
498            _ => Err(response_error(&self.url, &mut resp, "history")),
499        }
500    }
501}
502
503#[cfg(test)]
504mod tests {
505    use super::*;
506
507    #[test]
508    fn server_urls_keep_the_path_prefix() {
509        for base in ["https://h/dvs", "https://h/dvs/"] {
510            let url = Url::parse(base).unwrap();
511            let server = ServerBackend::new("project".into(), None, url.clone());
512            assert_eq!(
513                server.init_url().unwrap().as_str(),
514                "https://h/dvs/api/init"
515            );
516            assert_eq!(
517                server.access_url().unwrap().as_str(),
518                "https://h/dvs/api/project/access"
519            );
520            assert_eq!(
521                crate::auth::get_client_approve_url(&url, "ABCD")
522                    .unwrap()
523                    .as_str(),
524                "https://h/dvs/device/approve?user_code=ABCD"
525            );
526        }
527    }
528
529    #[test]
530    fn statuses_map_to_the_shared_codes() {
531        for (status, body, expected) in [
532            (401, "", Some(ErrorCode::AuthExpired)),
533            (403, "", Some(ErrorCode::AccessDenied)),
534            (
535                404,
536                "unknown project \"project\"",
537                Some(ErrorCode::ProjectNotFound),
538            ),
539            (404, "", Some(ErrorCode::ServerTooOld)),
540            (405, "", Some(ErrorCode::ServerTooOld)),
541            (503, "", Some(ErrorCode::ServerUnavailable)),
542            (409, "", None),
543        ] {
544            assert_eq!(http_status_code(status, body), expected, "{status} {body}");
545        }
546    }
547}
dvs/backends/server.rs

dvs/backends/
server.rs
