shep 0.12.2

The shep binary: a process manager that keeps a flock of long-running processes alive on macOS, Linux and Windows, with logs, watch and cron restarts, and webhook alerts
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
//! A bounded, redirect-refusing GET over `tokio-rustls`: [`get`], plus the
//! URL parsing ([`parse_url`]) and TLS setup ([`tls_connector`]) it shares
//! with `dog::bark::sinks`. An HTTP client, where `crate::http` is a
//! hand-rolled, TLS-free server. Either scheme is accepted; a caller
//! wanting only `https://` enforces that itself. A URL carrying
//! credentials before the host (`user@`, `user:pass@`) is refused.
//!
//! [`get`] refuses, in this order: a 3xx naming a `Location`; any other
//! non-2xx; any `Transfer-Encoding`, since it reads exactly
//! `Content-Length` bytes; a `Content-Length` absent, unparseable, or
//! contradicted by a second one; and one above the caller's `limit`,
//! rechecked as the body arrives. The head has its own cap,
//! [`MAX_HEADER_BYTES`]. No `Accept-Encoding` is sent.
use core::fmt;
use std::sync::Arc;
use std::time::Duration;

use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;
use tokio_rustls::rustls::pki_types::ServerName;

use crate::terminal_safe;

/// The most this module will read before the blank line that ends a
/// response's header block, status line included.
///
/// 64 KiB, within an order of magnitude of what nginx, Apache and Go's
/// `net/http` allow, and far more than any real response needs: the live
/// index answers in about 300 bytes of headers. Nothing else bounds a peer
/// that never sends a newline.
///
/// Not the caller's `limit`, which is a body budget.
pub const MAX_HEADER_BYTES: usize = 64 * 1024;

/// A URL, parsed into what [`get`] needs to reach it.
///
/// `Debug` is REDACTED (IR-41), and `path` is the field that needs it: a
/// Discord or Slack webhook URL is a bearer credential and carries its
/// token as a path segment, which is why
/// [`Sink`](crate::dog::bark::sinks::Sink) redacts its own `Debug` too. A
/// `Target` printed into a log, a panic message or an error chain must not
/// hand that token to whoever reads the log.
///
/// `host` needs no redaction: [`parse_url`] refuses a `user@` or
/// `user:pass@` authority rather than folding it into `host`.
#[derive(Clone, PartialEq, Eq)]
pub struct Target {
    /// `true` for `https://`, `false` for `http://`.
    pub https: bool,
    /// The host, without a port.
    pub host: String,
    /// The port: the URL's own, or the scheme's default (443/80).
    pub port: u16,
    /// The request path, always starting with `/`.
    pub path: String,
}

impl Target {
    /// The port a URL of this scheme reaches when it names none: 443 for
    /// `https://`, 80 for `http://`.
    ///
    /// One spelling for both readers of the rule: [`parse_url`], which
    /// assigns it, and [`build_get_request`], which decides from it whether
    /// the `Host` header names a port at all.
    const fn default_port(https: bool) -> u16 {
        if https { 443 } else { 80 }
    }
}

/// Manual: a derived `Debug` would print `path` in full, and `path` is
/// where a webhook's own credential lives. Every `Target` collapses to its
/// scheme, host and port, with `path` withheld.
impl fmt::Debug for Target {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "Target {{ https: {}, host: {:?}, port: {}, path: <redacted> }}",
            self.https, self.host, self.port
        )
    }
}

/// Why [`parse_url`] or [`get`] failed.
///
/// `Debug` needs no redaction: every field is a URL this module was given
/// or redirected to, an HTTP status, a byte count, or an OS error.
#[derive(Debug)]
pub enum FetchError {
    /// `url` did not parse as an absolute `http://`/`https://` URL. Carries
    /// a human-readable reason, never the raw bytes of a malformed input.
    ///
    /// The reason quotes `url` only through [`url_for_message`], which
    /// withholds anything holding an `@`. That is the backstop under
    /// [`url_carries_credentials`], which reads an authority and so can
    /// misread a url far enough off the grammar. An authority that really
    /// does carry `user@` or `user:pass@` gets its own reason,
    /// [`CREDENTIALS_REFUSAL`].
    Url(String),
    /// The connection failed, the TLS handshake failed, or the response
    /// was not well-formed HTTP: no parseable status line, a header block
    /// that never reached its terminating blank line, or a `Content-Length`
    /// that was missing, not a number, or contradicted by a second
    /// `Content-Length` header on the same response.
    Transport(std::io::Error),
    /// The response was outside 2xx (including a 3xx with no `Location` to
    /// report as a [`Self::Redirect`]).
    Status(u16),
    /// The response was a 3xx naming a `Location`, refused rather than
    /// followed.
    Redirect {
        /// The `Location` header's value, [`crate::terminal_safe::sanitise`]d
        /// at capture: a string the host chose that this `Display` prints
        /// to a terminal.
        location: String,
    },
    /// The response carried a `Transfer-Encoding` header. This client reads
    /// exactly `Content-Length` bytes, so a chunked body would be
    /// misparsed rather than decoded.
    Chunked,
    /// The declared or actual body size exceeded the caller's limit.
    TooLarge {
        /// The limit the caller passed to [`get`].
        limit: usize,
    },
    /// The status line and header block together ran past
    /// [`MAX_HEADER_BYTES`] without reaching the blank line that ends them.
    /// Separate from [`Self::TooLarge`], which is the caller's body budget.
    HeadersTooLarge {
        /// [`MAX_HEADER_BYTES`], named here so the message can state it.
        limit: usize,
    },
    /// `timeout` elapsed before the exchange finished.
    Timeout,
    /// The peer closed the connection before `Content-Length` bytes
    /// arrived.
    Truncated {
        /// The `Content-Length` the response declared.
        expected: usize,
        /// How many bytes actually arrived before the peer closed.
        got: usize,
    },
}

impl fmt::Display for FetchError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Url(reason) => write!(f, "not a fetchable url: {reason}"),
            Self::Transport(source) => write!(f, "fetch failed: {source}"),
            Self::Status(code) => write!(f, "fetch answered {code}"),
            Self::Redirect { location } => write!(f, "fetch was redirected to {location}"),
            Self::Chunked => {
                write!(
                    f,
                    "fetch response used transfer-encoding, which this client refuses to decode"
                )
            }
            Self::TooLarge { limit } => write!(f, "fetch response exceeded the {limit}-byte limit"),
            Self::HeadersTooLarge { limit } => {
                write!(f, "fetch response headers exceeded the {limit}-byte limit")
            }
            Self::Timeout => write!(f, "fetch timed out"),
            Self::Truncated { expected, got } => {
                write!(
                    f,
                    "fetch response was truncated: expected {expected} bytes, got {got}"
                )
            }
        }
    }
}

impl core::error::Error for FetchError {
    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
        match self {
            Self::Transport(source) => Some(source),
            Self::Url(_)
            | Self::Status(_)
            | Self::Redirect { .. }
            | Self::Chunked
            | Self::TooLarge { .. }
            | Self::HeadersTooLarge { .. }
            | Self::Timeout
            | Self::Truncated { .. } => None,
        }
    }
}

impl From<std::io::Error> for FetchError {
    fn from(source: std::io::Error) -> Self {
        Self::Transport(source)
    }
}

/// The reason [`parse_url`] and [`super::dog::bark::sinks`] both give for
/// a URL carrying credentials. It names no URL, which is the point: the
/// text being refused is the text holding the password.
pub const CREDENTIALS_REFUSAL: &str = concat!(
    "credentials before the host (`user@` or `user:pass@`) are not supported; ",
    "the url is not echoed, since it carries one"
);

/// How `url` may be named in a message: itself, or a fixed placeholder.
///
/// Deliberately blunter than [`url_carries_credentials`], and deliberately
/// the only rule any message consults. That predicate reads an authority,
/// so it can say precisely why a sink was refused; this one asks only
/// whether printing the text might print a secret, and an `@` anywhere is
/// enough to decline.
///
/// Two rules for one question drift, and did: [`parse_url`] withheld
/// `file:///etc/user:pw@host` on its own `@` test while
/// [`available_dogs`](crate::commands::query::available_dogs) printed that
/// same url in the sentence around it, because it asked the predicate
/// instead.
pub fn url_for_message(url: &str) -> &str {
    if url.contains('@') {
        "a url that may carry credentials"
    } else {
        url
    }
}

/// Where `rest`, an authority followed by whatever came after it, stops
/// being the authority.
///
/// `/`, `?` and `#` all end it. Splitting on `/` alone reads
/// `example.com?contact=alice@example.com` as one authority, which makes
/// [`url_carries_credentials`] call an `@` in a query a credential.
fn authority_of(rest: &str) -> &str {
    match rest.find(['/', '?', '#']) {
        Some(i) => &rest[..i],
        None => rest,
    }
}

/// Whether `url`'s authority carries a `user@` or `user:pass@` prefix.
///
/// The rule [`parse_url`] refuses on, separated out so a caller holding a
/// URL it has not parsed yet can refuse the same shape at config-load time
/// rather than at first use.
///
/// Deliberately blind to the scheme, and to whether there is one at all.
/// Keying this on `http://`/`https://` would have answered `false` for
/// `ftp://user:pass@host/` and for `HTTPS://user:pass@host/`, which
/// [`parse_url`] then refuses on the SCHEME instead, in a message that
/// quotes the whole url and hands the password back. A url this cannot
/// parse is exactly the one whose refusal must not echo it.
///
/// The cost is that `mailto:someone@example.com` reads as credentials.
/// It is refused either way, and a wrong reason on a url nothing here can
/// fetch is cheaper than a right one that prints a password.
pub fn url_carries_credentials(url: &str) -> bool {
    let rest = url.split_once("://").map_or(url, |(_scheme, rest)| rest);
    // A scheme-relative url has an authority with no scheme in front of
    // it, so there is no `://` to split on and the authority would
    // otherwise read as the empty string before the first `/`.
    let rest = rest.strip_prefix("//").unwrap_or(rest);
    authority_of(rest).contains('@')
}

/// Parses `url` into a [`Target`] naming where [`get`] should connect.
///
/// Hand-rolled: a fetch target is never more than a scheme, a host, an
/// optional port and a path.
///
/// Credentials before the host are refused rather than stripped: nothing
/// downstream of here sends an `Authorization` header, so a `user:pass@`
/// prefix could only ever be silently discarded or silently sent as part
/// of a hostname. `ProbeTarget::parse` refuses the same shape for the same
/// reason.
///
/// The authority ends at the first `/`, `?` or `#`. A query belongs to
/// the path from there; a fragment is dropped, being the client's own and
/// having no place in a request target.
///
/// # Errors
/// - [`FetchError::Url`] if `url` does not start with `http://` or
///   `https://`, carries a `user@` or `user:pass@` authority, names a
///   non-numeric port, or names no host. No such message quotes a `url`
///   holding an `@`; see [`url_for_message`].
pub fn parse_url(url: &str) -> Result<Target, FetchError> {
    // Ahead of the host/port split, which trusts the last colon: without
    // this, `user:pass@host:8443` parses to the host `user:pass@host` and
    // `user:pass@host` to the host `user` and the port `pass@host`, so a
    // password reaches either a `Target` field or, through that split's
    // own refusal, an error message quoting the whole URL.
    if url_carries_credentials(url) {
        return Err(FetchError::Url(CREDENTIALS_REFUSAL.to_owned()));
    }
    let (https, rest) = match url.strip_prefix("https://") {
        Some(rest) => (true, rest),
        None => match url.strip_prefix("http://") {
            Some(rest) => (false, rest),
            None => {
                return Err(FetchError::Url(format!(
                    "{} does not start with http:// or https://",
                    url_for_message(url)
                )));
            }
        },
    };
    let authority = authority_of(rest);
    let remainder = &rest[authority.len()..];
    // A fragment is the client's own: RFC 3986 gives it no meaning to a
    // server and RFC 7230's origin-form target has no field for it, so
    // carrying it into `path` would put it on the wire in the request
    // line. A query is the opposite and belongs there.
    let remainder = remainder.split('#').next().unwrap_or(remainder);
    // A `?` remainder has no leading `/` of its own and an origin-form
    // target needs one, as does an absent path.
    let path = match remainder.chars().next() {
        Some('/') => remainder.to_owned(),
        Some(_) => format!("/{remainder}"),
        None => "/".to_owned(),
    };
    let (host, port) = match authority.rsplit_once(':') {
        Some((h, p)) => (
            h,
            p.parse().map_err(|_err| {
                FetchError::Url(format!("{} has a non-numeric port", url_for_message(url)))
            })?,
        ),
        None => (authority, Target::default_port(https)),
    };
    if host.is_empty() {
        return Err(FetchError::Url(format!(
            "{} has no host",
            url_for_message(url)
        )));
    }
    Ok(Target {
        https,
        host: host.to_string(),
        port,
        path,
    })
}

/// The TLS connector every `https://` connection this crate makes shares,
/// built once on first use: a fresh one per connection would re-walk the
/// root store and re-derive the cipher suite set every time.
pub fn tls_connector() -> &'static tokio_rustls::TlsConnector {
    static CONNECTOR: std::sync::LazyLock<tokio_rustls::TlsConnector> =
        std::sync::LazyLock::new(|| {
            let roots = tokio_rustls::rustls::RootCertStore::from_iter(
                webpki_roots::TLS_SERVER_ROOTS.iter().cloned(),
            );
            let provider = Arc::new(tokio_rustls::rustls::crypto::ring::default_provider());
            let config = tokio_rustls::rustls::ClientConfig::builder_with_provider(provider)
                .with_safe_default_protocol_versions()
                .expect("ring's default cipher suites cover rustls's own default protocol versions")
                .with_root_certificates(roots)
                .with_no_client_auth();
            tokio_rustls::TlsConnector::from(Arc::new(config))
        });
    &CONNECTOR
}

/// Fetches `target` with a single GET, refusing anything but a plain 2xx
/// body no larger than `limit` bytes, bounded end to end by `timeout`.
///
/// # Errors
/// Every [`FetchError`] variant can come out of this call except
/// [`FetchError::Url`], which only [`parse_url`] produces. The module doc
/// lists the refusal order.
pub async fn get(target: &Target, limit: usize, timeout: Duration) -> Result<Vec<u8>, FetchError> {
    match tokio::time::timeout(timeout, get_inner(target, limit)).await {
        Ok(result) => result,
        Err(_elapsed) => Err(FetchError::Timeout),
    }
}

/// Connects to `target`, over TLS when `target.https`, and runs the
/// write/read exchange over whichever stream results.
async fn get_inner(target: &Target, limit: usize) -> Result<Vec<u8>, FetchError> {
    let request = build_get_request(target);
    let tcp = TcpStream::connect((target.host.as_str(), target.port)).await?;
    if target.https {
        let domain = ServerName::try_from(target.host.clone())
            .map_err(|source| FetchError::Transport(std::io::Error::other(source)))?;
        let tls = tls_connector()
            .connect(domain, tcp)
            .await
            .map_err(peer_transport_error)?;
        exchange(tls, &request, limit).await
    } else {
        exchange(tcp, &request, limit).await
    }
}

/// A transport failure the peer had a hand in wording, sanitised.
///
/// The OS writes most of these and is not the threat. A TLS handshake is
/// the exception: rustls's `CertificateError::NotValidForNameContext`
/// prints the names the peer's certificate presented, and [`FetchError`]'s
/// `Display` puts those in front of an operator.
///
/// Keeps the [`std::io::ErrorKind`] and drops the nested source, which
/// nothing walks.
fn peer_transport_error(source: std::io::Error) -> FetchError {
    FetchError::Transport(std::io::Error::new(
        source.kind(),
        terminal_safe::sanitise(&source.to_string()).0,
    ))
}

/// The request line and headers [`get_inner`] sends. `Host` names the port
/// only when it is off the scheme's own default (443/80), and there is no
/// `Accept-Encoding`.
fn build_get_request(target: &Target) -> String {
    let default_port = Target::default_port(target.https);
    let host = if target.port == default_port {
        target.host.clone()
    } else {
        format!("{}:{}", target.host, target.port)
    };
    format!(
        "GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n",
        path = target.path,
    )
}

/// Writes `request`, flushes, then reads back the response. The flush
/// matters on the TLS branch: `tokio-rustls` buffers writes in `rustls`'s
/// record layer, and skipping it leaves a request the peer never sees.
async fn exchange<S: AsyncRead + AsyncWrite + Unpin>(
    mut stream: S,
    request: &str,
    limit: usize,
) -> Result<Vec<u8>, FetchError> {
    stream.write_all(request.as_bytes()).await?;
    stream.flush().await?;
    read_response(stream, limit).await
}

/// Reads the status line, then headers to the blank line that ends them,
/// then exactly `Content-Length` bytes of body, refusing in the order the
/// module doc lists.
async fn read_response<S: AsyncRead + Unpin>(
    stream: S,
    limit: usize,
) -> Result<Vec<u8>, FetchError> {
    // One budget for the status line and every header line together, spent
    // through a `Take`: `read_line` itself is what is unbounded. Out of
    // budget it reports a clean zero, exactly as EOF does, and
    // `headers.limit()` is what tells the two apart below.
    let mut headers = BufReader::new(stream).take(MAX_HEADER_BYTES as u64);

    let mut status_line = String::new();
    headers.read_line(&mut status_line).await?;
    let code = parse_status_line(&status_line)?;

    let mut location: Option<String> = None;
    let mut transfer_encoding = false;
    let mut content_length: Option<u64> = None;
    // Set on an unparseable value or a second, disagreeing
    // `Content-Length`; the refusal is deferred to after the loop so it
    // keeps its place in the documented order.
    let mut content_length_ok = true;

    loop {
        let mut line = String::new();
        let read = headers.read_line(&mut line).await?;
        if read == 0 {
            if headers.limit() == 0 {
                return Err(FetchError::HeadersTooLarge {
                    limit: MAX_HEADER_BYTES,
                });
            }
            return Err(FetchError::Transport(std::io::Error::other(
                "response headers ended without a blank line",
            )));
        }
        if line == "\r\n" || line == "\n" {
            break;
        }
        let Some((name, value)) = line.split_once(':') else {
            return Err(FetchError::Transport(std::io::Error::other(
                "malformed header line: no colon",
            )));
        };
        let value = value.trim();
        match name.trim().to_ascii_lowercase().as_str() {
            // Sanitised at this one seam, where a header value becomes an
            // owned string this module keeps. Not hoisted over the whole
            // `match`: `content-length` is parsed below, and cleaning it
            // first would repair `1\u{200b}2` into a `12` this would honour.
            "location" => location = Some(terminal_safe::sanitise(value).0),
            "transfer-encoding" => transfer_encoding = true,
            "content-length" => match value.parse::<u64>() {
                Ok(parsed) => match content_length {
                    Some(existing) if existing != parsed => content_length_ok = false,
                    Some(_) => {}
                    None => content_length = Some(parsed),
                },
                Err(_err) => content_length_ok = false,
            },
            _ => {}
        }
    }

    if (300..400).contains(&code)
        && let Some(location) = location
    {
        return Err(FetchError::Redirect { location });
    }
    // A 3xx with no Location has nothing left to name.
    if !(200..300).contains(&code) {
        return Err(FetchError::Status(code));
    }
    if transfer_encoding {
        return Err(FetchError::Chunked);
    }
    if !content_length_ok {
        return Err(FetchError::Transport(std::io::Error::other(
            "content-length header was not a number, or two content-length headers disagreed",
        )));
    }
    let Some(content_length) = content_length else {
        return Err(FetchError::Transport(std::io::Error::other(
            "response carried no content-length",
        )));
    };
    if content_length > limit as u64 {
        return Err(FetchError::TooLarge { limit });
    }
    // Safe: just checked `content_length <= limit`, and `limit` is itself a
    // `usize`, so `content_length` fits in one.
    let expected = content_length as usize;

    // Anything the `BufReader` read past the blank line is still in its
    // buffer, so unwrapping the `Take` resumes where the header loop
    // stopped.
    let mut reader = headers.into_inner();

    let mut body = vec![0u8; expected];
    let mut filled = 0;
    while filled < expected {
        let read = reader.read(&mut body[filled..]).await?;
        if read == 0 {
            return Err(FetchError::Truncated {
                expected,
                got: filled,
            });
        }
        filled += read;
        // Unreachable in practice: `body` is sized to `expected`, which
        // `limit` already bounds, so `filled` cannot pass it here.
        if filled > limit {
            return Err(FetchError::TooLarge { limit });
        }
    }
    Ok(body)
}

/// The status code out of `status_line` (`"HTTP/1.1 200 OK\r\n"`), refusing
/// anything not shaped like an HTTP status line at all.
fn parse_status_line(status_line: &str) -> Result<u16, FetchError> {
    let mut parts = status_line.split_whitespace();
    match parts.next() {
        Some(version) if version.starts_with("HTTP/") => {}
        _ => {
            return Err(FetchError::Transport(std::io::Error::other(
                "response did not start with an HTTP status line",
            )));
        }
    }
    parts
        .next()
        .and_then(|code| code.parse().ok())
        .ok_or_else(|| FetchError::Transport(std::io::Error::other("malformed http status code")))
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Serves one canned response on an ephemeral port, then stops.
    async fn serve(response: &'static [u8]) -> Target {
        serve_owned(response.to_vec()).await
    }

    /// [`serve`] for a response a test builds at run time: the header-cap
    /// cases, whose responses are tens of kilobytes of padding.
    async fn serve_owned(response: Vec<u8>) -> Target {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let (mut stream, _peer) = listener.accept().await.unwrap();
            // Drains the request so the client's write never stalls on a
            // full socket buffer.
            let mut buf = [0u8; 1024];
            let _ = stream.read(&mut buf).await;
            let _ = stream.write_all(&response).await;
            let _ = stream.shutdown().await;
        });
        Target {
            https: false,
            host: "127.0.0.1".to_string(),
            port: addr.port(),
            path: "/".to_string(),
        }
    }

    /// Serves a response that opens a header and never ends it, for as
    /// long as anything is still reading.
    async fn serve_endless_header() -> Target {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let (mut stream, _peer) = listener.accept().await.unwrap();
            let mut buf = [0u8; 1024];
            let _ = stream.read(&mut buf).await;
            if stream
                .write_all(b"HTTP/1.1 200 OK\r\nX-Filler: ")
                .await
                .is_err()
            {
                return;
            }
            // Ends itself: a client that refuses closes the socket, and
            // the next write fails.
            let filler = vec![b'A'; 8 * 1024];
            while stream.write_all(&filler).await.is_ok() {}
        });
        Target {
            https: false,
            host: "127.0.0.1".to_string(),
            port: addr.port(),
            path: "/".to_string(),
        }
    }

    /// Asserts the exact cleaned string, not merely the absence of an
    /// escape: the redirect must still name where it pointed.
    #[tokio::test]
    async fn a_hostile_location_header_cannot_drive_the_terminal_it_prints_to() {
        let target = serve(
            b"HTTP/1.1 302 Found\r\nLocation: \x1b[2J\x1b]0;pwned\x07/gone\r\nContent-Length: 0\r\n\r\n",
        )
        .await;
        let err = get(&target, 1 << 20, Duration::from_secs(5))
            .await
            .expect_err("refused");
        let FetchError::Redirect { location } = &err else {
            panic!("wrong variant: {err:?}")
        };
        assert_eq!(location, "[2J]0;pwned/gone", "location was not sanitised");
        assert!(
            !err.to_string().chars().any(char::is_control),
            "a control character reached the message: {:?}",
            err.to_string()
        );
    }

    /// Every response here is hostile in a different place and drives a
    /// different refusal, so a future variant that starts carrying
    /// response-derived text has to pass this to ship.
    #[tokio::test]
    async fn no_refusal_hands_a_terminal_a_character_the_host_chose() {
        let hostile: [(&'static [u8], &str); 6] = [
            (
                b"HTTP/1.1 302 Found\r\nLocation: \x1b[2J\x07\r\nContent-Length: 0\r\n\r\n",
                "a redirect naming a hostile location",
            ),
            (
                b"HTTP/1.1 404 \x1b[2JNot Found\r\nContent-Length: 0\r\n\r\n",
                "a non-2xx whose reason phrase is hostile",
            ),
            (
                b"HTTP/1.1 200 OK\r\nTransfer-Encoding: \x1b[2Jchunked\r\n\r\n0\r\n\r\n",
                "a chunked refusal whose header value is hostile",
            ),
            (
                b"HTTP/1.1 200 OK\r\nContent-Length: \x1b[2Jnope\r\n\r\n",
                "an unparseable content-length that is hostile",
            ),
            (
                b"\x1b[2J\x07 NOT HTTP AT ALL\r\n\r\n",
                "a status line that is not HTTP and is hostile",
            ),
            (
                b"HTTP/1.1 200 OK\r\n\x1b[2Jno-colon-here\r\n\r\n",
                "a header line with no colon, hostile",
            ),
        ];
        for (response, why) in hostile {
            let target = serve(response).await;
            let err = get(&target, 1 << 20, Duration::from_secs(5))
                .await
                .expect_err(why);
            let shown = err.to_string();
            assert!(
                !shown.chars().any(char::is_control),
                "{why}: a control character reached the message: {shown:?}"
            );
        }
    }

    /// The two-second budget is the forcing mechanism: a bounded refusal
    /// comes back in milliseconds.
    #[tokio::test]
    async fn a_header_that_never_ends_is_refused_rather_than_read_forever() {
        let target = serve_endless_header().await;
        let err = get(&target, 1 << 20, Duration::from_secs(2))
            .await
            .expect_err("refused");
        assert!(
            matches!(
                err,
                FetchError::HeadersTooLarge {
                    limit: MAX_HEADER_BYTES
                }
            ),
            "{err:?}"
        );
    }

    #[tokio::test]
    async fn a_header_block_just_under_the_cap_is_still_read() {
        let tail = b"\r\nContent-Length: 5\r\n\r\nhello";
        let head = b"HTTP/1.1 200 OK\r\nX-Pad: ";
        let pad = MAX_HEADER_BYTES - head.len() - tail.len();
        let mut response = head.to_vec();
        response.extend(std::iter::repeat_n(b'A', pad));
        response.extend_from_slice(tail);
        let target = serve_owned(response).await;
        let body = get(&target, 1 << 20, Duration::from_secs(5))
            .await
            .expect("read");
        assert_eq!(body, b"hello");
    }

    #[tokio::test]
    async fn a_content_length_body_is_read_exactly() {
        let target = serve(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello").await;
        let body = get(&target, 1 << 20, Duration::from_secs(5))
            .await
            .expect("read");
        assert_eq!(body, b"hello");
    }

    #[tokio::test]
    async fn a_chunked_response_is_refused_rather_than_misparsed() {
        let target =
            serve(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n")
                .await;
        assert!(matches!(
            get(&target, 1 << 20, Duration::from_secs(5)).await,
            Err(FetchError::Chunked)
        ));
    }

    #[tokio::test]
    async fn a_redirect_is_refused_and_names_where_it_pointed() {
        let target = serve(
            b"HTTP/1.1 301 Moved\r\nLocation: https://elsewhere/\r\nContent-Length: 0\r\n\r\n",
        )
        .await;
        let err = get(&target, 1 << 20, Duration::from_secs(5))
            .await
            .expect_err("refused");
        let FetchError::Redirect { location } = err else {
            panic!("wrong variant: {err:?}")
        };
        assert_eq!(location, "https://elsewhere/");
    }

    #[tokio::test]
    async fn a_body_over_the_limit_is_refused_before_it_is_read() {
        let target = serve(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n").await;
        assert!(matches!(
            get(&target, 10, Duration::from_secs(5)).await,
            Err(FetchError::TooLarge { limit: 10 })
        ));
    }

    #[tokio::test]
    async fn a_non_2xx_carries_its_status() {
        let target = serve(b"HTTP/1.1 500 Oops\r\nContent-Length: 0\r\n\r\n").await;
        assert!(matches!(
            get(&target, 1 << 20, Duration::from_secs(5)).await,
            Err(FetchError::Status(500))
        ));
    }

    #[tokio::test]
    async fn a_peer_that_closes_mid_body_is_an_error_not_a_short_read() {
        let target = serve(b"HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nshort").await;
        let err = get(&target, 1 << 20, Duration::from_secs(5))
            .await
            .expect_err("refused");
        assert!(
            matches!(
                err,
                FetchError::Truncated {
                    expected: 10,
                    got: 5
                }
            ),
            "{err:?}"
        );
    }

    #[tokio::test]
    async fn a_missing_content_length_is_refused() {
        let target = serve(b"HTTP/1.1 200 OK\r\n\r\nbody").await;
        assert!(get(&target, 1 << 20, Duration::from_secs(5)).await.is_err());
    }

    #[test]
    fn a_url_that_is_not_http_or_https_is_refused() {
        assert!(matches!(
            parse_url("file:///etc/passwd"),
            Err(FetchError::Url(_))
        ));
        assert!(matches!(parse_url("not a url"), Err(FetchError::Url(_))));
    }

    /// Both halves matter. The refusal is the fix; the exact string is
    /// what proves the refusal did not itself print the password, which is
    /// what quoting `url` in the message would have done.
    #[test]
    fn a_url_carrying_credentials_is_refused_without_echoing_them() {
        for url in [
            "https://user:hunter2@example.com:8443/webhook",
            "https://user:hunter2@example.com/webhook",
            "http://hunter2@example.com/webhook",
            // Schemes this module cannot fetch at all. Each used to be
            // refused on the scheme, in a message quoting the whole url.
            "ftp://user:hunter2@example.com/webhook",
            "HTTPS://user:hunter2@example.com/webhook",
            "user:hunter2@example.com/webhook",
            // Scheme-relative: an authority with nothing in front of it,
            // so there is no `://` to find and the leading `//` has to be
            // stepped over or the authority reads as empty.
            "//user:hunter2@example.com/webhook",
        ] {
            let err = parse_url(url).unwrap_err();
            assert!(matches!(err, FetchError::Url(_)), "{url}: {err:?}");
            assert_eq!(
                err.to_string(),
                "not a fetchable url: credentials before the host (`user@` or `user:pass@`) are \
                 not supported; the url is not echoed, since it carries one",
                "{url}"
            );
            assert!(!format!("{err} {err:?}").contains("hunter2"), "{url}");
        }
    }

    /// The backstop under [`url_carries_credentials`]. That predicate
    /// reads an authority, and a url far enough off the grammar is one it
    /// can misread: in each of these the `@` is in a path, so the
    /// predicate says `false` and some other refusal is what sees the url.
    /// Every refusal names the url through [`url_for_message`] rather than
    /// trusting the predicate to have been right about where the authority
    /// ended.
    #[test]
    fn no_refusal_quotes_a_url_holding_an_at_sign() {
        for url in [
            // Refused on the scheme.
            "file:///etc/pass@wd",
            // Refused on the port, with the `@` off in the path.
            "https://example.com:notaport/etc/pass@wd",
        ] {
            assert!(!url_carries_credentials(url), "{url}");
            let err = parse_url(url).unwrap_err();
            let rendered = format!("{err} {err:?}");
            assert!(!rendered.contains("pass@wd"), "{url}: {rendered}");
            assert!(
                rendered.contains("a url that may carry credentials"),
                "{url}: {rendered}"
            );
        }
    }

    /// Neither a query nor a fragment is part of the host. Reading them as
    /// authority made an `@` in a query look like a credential, so
    /// `?contact=alice@example.com` was refused as one.
    ///
    /// They part company after that: a query is the request target's, a
    /// fragment is the client's and is dropped.
    #[test]
    fn a_query_joins_the_path_and_a_fragment_is_dropped() {
        for (url, path) in [
            (
                "https://example.com?contact=alice@example.com",
                "/?contact=alice@example.com",
            ),
            ("https://example.com#a@b", "/"),
            ("https://example.com/hook#frag", "/hook"),
            ("https://example.com/hook?a=b#frag", "/hook?a=b"),
            ("https://example.com/hook?a=b", "/hook?a=b"),
            ("https://example.com", "/"),
        ] {
            let target = parse_url(url).unwrap_or_else(|err| panic!("{url}: {err}"));
            assert_eq!(target.host, "example.com", "{url}");
            assert_eq!(target.port, 443, "{url}");
            assert_eq!(target.path, path, "{url}");
        }
        assert!(!url_carries_credentials(
            "https://example.com?contact=alice@example.com"
        ));
    }

    /// The door the fragment would have gone out of. `parse_url` dropping
    /// it is the fix; this is the assertion that nothing downstream puts
    /// it back.
    #[test]
    fn no_fragment_reaches_the_request_line() {
        let target = parse_url("https://example.com/hook?a=b#sentinelfragment")
            .expect("a url with a query and a fragment parses");
        let request = build_get_request(&target);
        assert!(
            request.starts_with("GET /hook?a=b HTTP/1.1\r\n"),
            "{request}"
        );
        assert!(!request.contains("sentinelfragment"), "{request}");
    }

    /// A Discord webhook URL is a bearer credential and carries its token
    /// as a path segment, so `Target`'s `Debug` withholds the path the way
    /// `Sink`'s withholds the whole URL.
    #[test]
    fn a_targets_debug_redacts_the_path() {
        let target = parse_url("https://discord.com/api/webhooks/123/s3cr3t-token")
            .expect("a plain https url parses");
        let rendered = format!("{target:?}");
        assert_eq!(
            rendered,
            r#"Target { https: true, host: "discord.com", port: 443, path: <redacted> }"#
        );
        assert!(!rendered.contains("s3cr3t-token"));
    }

    /// A smuggling-style ambiguity: picking one would let a proxy make
    /// this client and whatever is downstream of it disagree about where
    /// the body ends.
    #[tokio::test]
    async fn two_disagreeing_content_lengths_are_refused() {
        let target =
            serve(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nContent-Length: 6\r\n\r\nhello!").await;
        assert!(get(&target, 1 << 20, Duration::from_secs(5)).await.is_err());
    }

    #[tokio::test]
    async fn a_status_line_that_is_not_http_is_refused() {
        let target = serve(b"NOT HTTP AT ALL\r\n\r\n").await;
        assert!(get(&target, 1 << 20, Duration::from_secs(5)).await.is_err());
    }

    /// An EOF here must not read as a zero-length body.
    #[tokio::test]
    async fn headers_with_no_terminating_blank_line_are_refused() {
        let target = serve(b"HTTP/1.1 200 OK\r\nContent-Length: 5").await;
        assert!(get(&target, 1 << 20, Duration::from_secs(5)).await.is_err());
    }

    /// Refused as an ordinary [`FetchError::Status`], having nowhere to
    /// point a caller.
    #[tokio::test]
    async fn a_redirect_with_no_location_is_still_refused() {
        let target = serve(b"HTTP/1.1 302 Found\r\nContent-Length: 0\r\n\r\n").await;
        assert!(matches!(
            get(&target, 1 << 20, Duration::from_secs(5)).await,
            Err(FetchError::Status(302))
        ));
    }
}