flagon-io/g1t

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Merge branch 'main' into rm-agents12h ago		1	//! Brokered secrets on a self-hosted runner. A brokered secret's value
		2	//! never reaches the computer a job runs on: the job has a stand-in
		3	//! (`g1tsec_…`) in its place, and g1t fills the value in on the job's
		4	//! requests to the secret's hosts. On g1t's own machines every request
		5	//! leaves through g1t; here, this harness sends the ones that need it
		6	//! there itself.
		7	//!
		8	//! The job's spec lists its stand-ins and their hosts (`brokered`; never a
		9	//! value), and its steps get `HTTPS_PROXY` pointing at a proxy this
		10	//! harness runs on 127.0.0.1 for the job's life:
		11	//!
		12	//! - **A connection to any other host** is a plain tunnel: the proxy never
		13	//! sees inside it.
		14	//! - **A connection to a brokered host** is opened with a certificate made
		15	//! for that host by a certificate authority made for this job alone,
		16	//! which the steps' tools are pointed at (`SSL_CERT_FILE` and the
		17	//! others, as g1t's own sandboxes set them), and which can sign for the
		18	//! brokered hosts only (a name constraint). Its key is never written
		19	//! down. Each request on it that carries one of the job's stand-ins is
		20	//! passed to g1t, `POST /actions/jobs/{job}/broker` with the job's token
		21	//! (services/runner broker-relay.ts), which fills the value in, sends
		22	//! it, and answers with what came back, the value swapped back for its
		23	//! stand-in. A request without a stand-in goes straight to the host.
		24	//!
		25	//! So the harness, the job and the computer only ever hold stand-ins. Not
		26	//! on Windows yet, whose tools trust the system's certificate store rather
		27	//! than these variables; there, as with a harness from before this, the
		28	//! stand-in goes as it is and the job's log says it will not be filled in.
		29
		30	use std::collections::{BTreeMap, HashMap};
		31	use std::io::{self, BufRead, BufReader, Read, Write};
		32	use std::net::{Shutdown, TcpListener, TcpStream, ToSocketAddrs};
		33	use std::path::{Path, PathBuf};
		34	use std::sync::atomic::{AtomicUsize, Ordering};
		35	use std::sync::{Arc, Mutex};
		36	use std::time::Duration;
		37
		38	use anyhow::{Context, Result, anyhow};
		39	use base64::Engine;
		40	use base64::engine::general_purpose::STANDARD as BASE64;
		41	use rustls::pki_types::pem::PemObject;
		42	use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName};
		43	use rustls::{ClientConfig, ClientConnection, RootCertStore, ServerConfig, ServerConnection, StreamOwned};
		44	use serde_json::{Value, json};
		45
		46	use super::report::Api;
		47
		48	/// How a stand-in starts (`STAND_IN_PREFIX` in g1t's contracts).
		49	pub(crate) const STAND_IN_PREFIX: &str = "g1tsec_";
		50	/// The largest request body passed to g1t, as broker-relay.ts takes it.
		51	pub(crate) const MAX_BODY_BYTES: usize = 32 * 1024 * 1024;
		52	/// The longest request head read.
		53	const MAX_HEAD_BYTES: usize = 64 * 1024;
		54	/// How long a connection may sit idle before the proxy lets it go.
		55	const IDLE: Duration = Duration::from_secs(5 * 60);
		56	/// How long one request passed to g1t may take, its answer included.
		57	const RELAY_TIME: Duration = Duration::from_secs(10 * 60);
		58	const CONNECT_TIME: Duration = Duration::from_secs(30);
		59	/// The most connections open through the proxy at once.
		60	const MAX_CONNECTIONS: usize = 256;
		61	/// The request to make, and the answer, as broker-relay.ts names them.
		62	const REQUEST_HEADER: &str = "x-g1t-broker-request";
		63	const ANSWER_HEADER: &str = "x-g1t-broker-response";
		64	/// Headers about one connection rather than the request, which are not
		65	/// passed on either way.
		66	const HOP_BY_HOP: &[&str] = &[
		67	"connection",
		68	"keep-alive",
		69	"proxy-authenticate",
		70	"proxy-authorization",
		71	"proxy-connection",
		72	"te",
		73	"trailer",
		74	"transfer-encoding",
		75	"upgrade",
		76	"content-length",
		77	];
		78	/// The variables the job's tools read their trusted certificates from,
		79	/// pointed at the system's bundle with this job's authority added, as g1t's
		80	/// own sandboxes set them (services/runner egress.ts, docker/oci.rs).
		81	/// `NODE_EXTRA_CA_CERTS` takes the authority alone.
		82	const BUNDLE_VARIABLES: &[&str] = &["SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE", "PIP_CERT", "GIT_SSL_CAINFO", "CARGO_HTTP_CAINFO"];
		83	/// Where a system keeps its bundle of trusted certificates: Debian and
		84	/// Ubuntu, Fedora and RHEL, Alpine and macOS, openSUSE.
		85	const SYSTEM_BUNDLES: &[&str] =
		86	&["/etc/ssl/certs/ca-certificates.crt", "/etc/pki/tls/certs/ca-bundle.crt", "/etc/ssl/cert.pem", "/etc/ssl/ca-bundle.pem"];
		87
		88	/// A stand-in the job was given, with its secret's key and hosts.
		89	#[derive(Clone, Debug, PartialEq)]
		90	pub(crate) struct StandIn {
		91	pub(crate) stand_in: String,
		92	pub(crate) name: String,
		93	pub(crate) hosts: Vec<String>,
		94	}
		95
		96	/// The job's stand-ins, from its spec's `brokered`: none from a g1t that
		97	/// does not broker for this runner.
		98	pub(crate) fn stand_ins(spec: &Value) -> Vec<StandIn> {
		99	let Some(entries) = spec.get("brokered").and_then(Value::as_array) else { return Vec::new() };
		100	entries
		101	.iter()
		102	.filter_map(|entry| {
		103	let stand_in = entry.get("stand_in").or_else(|| entry.get("standIn"))?.as_str()?;
		104	let hosts: Vec<String> = entry.get("hosts")?.as_array()?.iter().filter_map(Value::as_str).map(str::to_owned).collect();
		105	(stand_in.starts_with(STAND_IN_PREFIX) && !hosts.is_empty()).then(|| StandIn {
		106	stand_in: stand_in.to_owned(),
		107	name: entry.get("name").and_then(Value::as_str).unwrap_or_default().to_owned(),
		108	hosts,
		109	})
		110	})
		111	.collect()
		112	}
		113
		114	/// Whether this harness can broker its job's requests: on Linux or macOS,
		115	/// and when the computer has no proxy of its own set for HTTPS, which this
		116	/// one would have to chain to.
		117	pub(crate) fn possible(env: impl Fn(&str) -> Option<String>) -> bool {
		118	cfg!(unix) && ["HTTPS_PROXY", "https_proxy", "ALL_PROXY", "all_proxy"].iter().all(|name| !matches!(env(name), Some(value) if !value.trim().is_empty()))
		119	}
		120
		121	/// A host as a list names it: lowercase, without its port or a final dot.
		122	pub(crate) fn normalize_host(host: &str) -> String {
		123	let name = host.trim().to_ascii_lowercase();
		124	if name.starts_with('[') {
		125	return name.find(']').map_or(name.clone(), |end| name[..=end].to_owned());
		126	}
		127	let name = name.split(':').next().unwrap_or_default();
		128	name.trim_end_matches('.').to_owned()
		129	}
		130
		131	/// Whether `host` is on the list, as g1t reads a brokered secret's hosts
		132	/// (services/runner egress.ts `allows`): `example.com` is that host alone,
		133	/// `*.example.com` its subdomains at any depth but not itself.
		134	pub(crate) fn allows(patterns: &[String], host: &str) -> bool {
		135	let name = normalize_host(host);
		136	!name.is_empty()
		137	&& patterns.iter().any(|raw| {
		138	let pattern = raw.trim().to_ascii_lowercase();
		139	match pattern.strip_prefix('*') {
		140	Some(suffix) if suffix.starts_with('.') => name.len() > suffix.len() && name.ends_with(suffix),
		141	_ => pattern == name,
		142	}
		143	})
		144	}
		145
		146	/// A header's value as a stand-in may sit in it: Basic credentials decoded.
		147	fn readable(value: &str) -> Option<String> {
		148	let (scheme, rest) = value.trim().split_once(' ')?;
		149	if !scheme.eq_ignore_ascii_case("basic") {
		150	return None;
		151	}
		152	String::from_utf8(BASE64.decode(rest.trim()).ok()?).ok()
		153	}
		154
		155	/// Whether a request carries one of the job's stand-ins: in its target, a
		156	/// header, or Basic credentials, as broker.ts finds them.
		157	pub(crate) fn carries(target: &str, headers: &[(String, String)], stand_ins: &[StandIn]) -> bool {
		158	stand_ins.iter().any(|known| {
		159	target.contains(&known.stand_in)
		160	|| headers.iter().any(|(_, value)| value.contains(&known.stand_in) || readable(value).is_some_and(|decoded| decoded.contains(&known.stand_in)))
		161	})
		162	}
		163
		164	fn hop_by_hop(name: &str) -> bool {
		165	HOP_BY_HOP.iter().any(|hop| hop.eq_ignore_ascii_case(name))
		166	}
		167
		168	// --- HTTP/1.1, as much as the proxy reads ----------------------------------
		169
		170	/// A request's line and headers.
		171	#[derive(Debug, PartialEq)]
		172	pub(crate) struct Head {
		173	pub(crate) method: String,
		174	pub(crate) target: String,
		175	pub(crate) version: String,
		176	pub(crate) headers: Vec<(String, String)>,
		177	}
		178
		179	impl Head {
		180	fn header(&self, name: &str) -> Option<&str> {
		181	self.headers.iter().find(|(n, _)| n.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())
		182	}
		183
		184	/// Whether the client wants the connection closed after this request.
		185	fn closes(&self) -> bool {
		186	match self.header("connection") {
		187	Some(value) => value.split(',').any(|token| token.trim().eq_ignore_ascii_case("close")),
		188	None => self.version.eq_ignore_ascii_case("HTTP/1.0"),
		189	}
		190	}
		191	}
		192
		193	/// The error a body past [`MAX_BODY_BYTES`] reads as.
		194	#[derive(Debug)]
		195	struct TooLarge;
		196
		197	impl std::fmt::Display for TooLarge {
		198	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		199	write!(f, "the request's body is larger than {} MB", MAX_BODY_BYTES / 1024 / 1024)
		200	}
		201	}
		202
		203	impl std::error::Error for TooLarge {}
		204
		205	fn too_large(error: &io::Error) -> bool {
		206	error.get_ref().is_some_and(|inner| inner.is::<TooLarge>())
		207	}
		208
		209	fn invalid(message: &str) -> io::Error {
		210	io::Error::new(io::ErrorKind::InvalidData, message.to_owned())
		211	}
		212
		213	/// One line, at most `limit` bytes, without its line ending; None at the end.
		214	fn read_line(reader: &mut impl BufRead, limit: usize) -> io::Result<Option<String>> {
		215	let mut line = Vec::new();
		216	let read = reader.take(limit as u64 + 1).read_until(b'\n', &mut line)?;
		217	if read == 0 {
		218	return Ok(None);
		219	}
		220	if line.len() > limit {
		221	return Err(invalid("a line of the request is too long"));
		222	}
		223	while matches!(line.last(), Some(b'\n' | b'\r')) {
		224	line.pop();
		225	}
		226	Ok(Some(String::from_utf8_lossy(&line).into_owned()))
		227	}
		228
		229	/// A request's head, or None where the connection ends between requests.
		230	pub(crate) fn read_head(reader: &mut impl BufRead) -> io::Result<Option<Head>> {
		231	let mut lines: Vec<String> = Vec::new();
		232	let mut total = 0;
		233	loop {
		234	let Some(line) = read_line(reader, MAX_HEAD_BYTES)? else {
		235	return if lines.is_empty() { Ok(None) } else { Err(io::ErrorKind::UnexpectedEof.into()) };
		236	};
		237	total += line.len() + 2;
		238	if total > MAX_HEAD_BYTES {
		239	return Err(invalid("the request's head is too long"));
		240	}
		241	if line.is_empty() {
		242	// Blank lines before a request are allowed, and skipped.
		243	if lines.is_empty() {
		244	continue;
		245	}
		246	break;
		247	}
		248	lines.push(line);
		249	}
		250	let mut parts = lines[0].split_whitespace();
		251	let (Some(method), Some(target), Some(version)) = (parts.next(), parts.next(), parts.next()) else {
		252	return Err(invalid("the request line does not read"));
		253	};
		254	let headers = lines[1..]
		255	.iter()
		256	.map(|line| line.split_once(':').map(|(name, value)| (name.trim().to_owned(), value.trim().to_owned())).ok_or_else(|| invalid("a header does not read")))
		257	.collect::<io::Result<Vec<_>>>()?;
		258	Ok(Some(Head { method: method.to_owned(), target: target.to_owned(), version: version.to_owned(), headers }))
		259	}
		260
		261	/// A request's body, by its length or in chunks, at most [`MAX_BODY_BYTES`].
		262	pub(crate) fn read_body(reader: &mut impl BufRead, head: &Head) -> io::Result<Vec<u8>> {
		263	if head.header("transfer-encoding").is_some_and(|te| te.to_ascii_lowercase().contains("chunked")) {
		264	let mut body = Vec::new();
		265	loop {
		266	let line = read_line(reader, 1024)?.ok_or(io::ErrorKind::UnexpectedEof)?;
		267	let size = usize::from_str_radix(line.split(';').next().unwrap_or_default().trim(), 16).map_err(|_| invalid("a chunk's size does not read"))?;
		268	if size == 0 {
		269	// Trailers, which go no further, up to the blank line.
		270	while !read_line(reader, MAX_HEAD_BYTES)?.ok_or(io::ErrorKind::UnexpectedEof)?.is_empty() {}
		271	return Ok(body);
		272	}
		273	if body.len() + size > MAX_BODY_BYTES {
		274	return Err(io::Error::other(TooLarge));
		275	}
		276	let start = body.len();
		277	body.resize(start + size, 0);
		278	reader.read_exact(&mut body[start..])?;
		279	read_line(reader, 2)?;
		280	}
		281	}
		282	let length = match head.header("content-length") {
		283	Some(value) => value.trim().parse::<usize>().map_err(|_| invalid("the content length does not read"))?,
		284	None => 0,
		285	};
		286	if length > MAX_BODY_BYTES {
		287	return Err(io::Error::other(TooLarge));
		288	}
		289	let mut body = vec![0; length];
		290	reader.read_exact(&mut body)?;
		291	Ok(body)
		292	}
		293
		294	fn reason(status: u16) -> &'static str {
		295	match status {
		296	200 => "OK",
		297	201 => "Created",
		298	202 => "Accepted",
		299	204 => "No Content",
		300	206 => "Partial Content",
		301	301 => "Moved Permanently",
		302	302 => "Found",
		303	303 => "See Other",
		304	304 => "Not Modified",
		305	307 => "Temporary Redirect",
		306	308 => "Permanent Redirect",
		307	400 => "Bad Request",
		308	401 => "Unauthorized",
		309	403 => "Forbidden",
		310	404 => "Not Found",
		311	405 => "Method Not Allowed",
		312	409 => "Conflict",
		313	413 => "Content Too Large",
		314	422 => "Unprocessable Content",
		315	429 => "Too Many Requests",
		316	500 => "Internal Server Error",
		317	502 => "Bad Gateway",
		318	503 => "Service Unavailable",
		319	504 => "Gateway Timeout",
		320	_ => "Status",
		321	}
		322	}
		323
		324	/// A short answer of the proxy's own, in plain text.
		325	fn write_plain(out: &mut impl Write, status: u16, message: &str) -> io::Result<()> {
		326	let body = format!("g1t: {message}\n");
		327	write!(
		328	out,
		329	"HTTP/1.1 {status} {}\r\ncontent-type: text/plain; charset=utf-8\r\ncontent-length: {}\r\nx-g1t-broker: refused\r\n\r\n{body}",
		330	reason(status),
		331	body.len()
		332	)?;
		333	out.flush()
		334	}
		335
		336	// --- g1t's side --------------------------------------------------------------
		337
		338	/// What g1t answered a request passed to it.
		339	pub(crate) enum Answer {
		340	/// The host's answer, scrubbed: its status, headers and body.
		341	Relayed { status: u16, headers: Vec<(String, String)>, body: Box<dyn Read + Send> },
		342	/// g1t's own refusal, or that it could not be reached.
		343	Refused { status: u16, message: String },
		344	}
		345
		346	/// Passes requests to g1t's broker with the job's token.
		347	#[derive(Clone)]
		348	pub(crate) struct Relay {
		349	url: String,
		350	token: String,
		351	agent: ureq::Agent,
		352	}
		353
		354	impl Relay {
		355	/// For the job `api` is for, with its current token (taken after the
		356	/// spec was fetched, so the one it started with is already spent).
		357	pub(crate) fn new(api: &Api) -> Relay {
		358	Relay {
		359	url: format!("{}/actions/jobs/{}/broker", api.base, api.job),
		360	token: api.token.clone(),
		361	// Never a redirect: g1t answers itself.
		362	agent: ureq::AgentBuilder::new().redirects(0).timeout(RELAY_TIME).build(),
		363	}
		364	}
		365
		366	/// Passes one request to g1t: the token in a header, never the URL.
		367	pub(crate) fn send(&self, method: &str, url: &str, headers: &[(String, String)], body: &[u8]) -> Answer {
		368	let wanted = json!({ "method": method, "url": url, "headers": headers.iter().map(|(n, v)| [n, v]).collect::<Vec<_>>() });
		369	let sent = self
		370	.agent
		371	.post(&self.url)
		372	.set("authorization", &format!("Bearer {}", self.token))
		373	.set(REQUEST_HEADER, &BASE64.encode(wanted.to_string()))
		374	.set("content-type", "application/octet-stream")
		375	.send_bytes(body);
		376	match sent {
		377	Ok(response) => {
		378	let answer = response
		379	.header(ANSWER_HEADER)
		380	.and_then(|header| BASE64.decode(header.trim()).ok())
		381	.and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok());
		382	let Some(answer) = answer else {
		383	return Answer::Refused { status: 502, message: "g1t's answer did not read.".into() };
		384	};
		385	let status = answer["status"].as_u64().and_then(|s| u16::try_from(s).ok()).filter(|s| (100..600).contains(s)).unwrap_or(502);
		386	let headers = answer["headers"]
		387	.as_array()
		388	.map(|pairs| {
		389	pairs
		390	.iter()
		391	.filter_map(|pair| Some((pair.get(0)?.as_str()?.to_owned(), pair.get(1)?.as_str()?.to_owned())))
		392	.collect()
		393	})
		394	.unwrap_or_default();
		395	Answer::Relayed { status, headers, body: Box::new(response.into_reader()) }
		396	}
		397	Err(ureq::Error::Status(status, response)) => {
		398	let message = response
		399	.into_json::<Value>()
		400	.ok()
		401	.and_then(|body| body["error"]["message"].as_str().map(str::to_owned))
		402	.unwrap_or_else(|| format!("g1t's broker answered {status}."));
		403	// A refusal of g1t's is the job's to see; anything else
		404	// (g1t's own trouble, a spent token) is a bad gateway.
		405	let status = if matches!(status, 403 | 413) { status } else { 502 };
		406	Answer::Refused { status, message }
		407	}
		408	Err(error) => Answer::Refused { status: 502, message: format!("g1t's broker could not be reached: {error}") },
		409	}
		410	}
		411	}
		412
		413	/// Writes g1t's answer to the client, its body in chunks.
		414	fn write_answer(out: &mut impl Write, method: &str, answer: Answer) -> io::Result<()> {
		415	let (status, headers, mut body) = match answer {
		416	Answer::Refused { status, message } => return write_plain(out, status, &message),
		417	Answer::Relayed { status, headers, body } => (status, headers, body),
		418	};
		419	let mut head = format!("HTTP/1.1 {status} {}\r\n", reason(status));
		420	for (name, value) in headers.iter().filter(|(name, _)| !hop_by_hop(name)) {
		421	// A header may not break the answer's framing.
		422	if !name.contains(['\r', '\n', ':']) && !value.contains(['\r', '\n']) {
		423	head.push_str(&format!("{name}: {value}\r\n"));
		424	}
		425	}
		426	let bodiless = method.eq_ignore_ascii_case("HEAD") || status < 200 || status == 204 || status == 304;
		427	if bodiless {
		428	head.push_str("\r\n");
		429	out.write_all(head.as_bytes())?;
		430	return out.flush();
		431	}
		432	head.push_str("transfer-encoding: chunked\r\n\r\n");
		433	out.write_all(head.as_bytes())?;
		434	let mut buffer = vec![0; 64 * 1024];
		435	loop {
		436	let read = body.read(&mut buffer)?;
		437	if read == 0 {
		438	break;
		439	}
		440	write!(out, "{read:x}\r\n")?;
		441	out.write_all(&buffer[..read])?;
		442	out.write_all(b"\r\n")?;
		443	}
		444	out.write_all(b"0\r\n\r\n")?;
		445	out.flush()
		446	}
		447
		448	// --- Certificates --------------------------------------------------------------
		449
		450	/// The job's own certificate authority, and the certificates it made.
		451	pub(crate) struct Authority {
		452	cert: rcgen::Certificate,
		453	key: rcgen::KeyPair,
		454	valid: (time::OffsetDateTime, time::OffsetDateTime),
		455	/// Each host's TLS configuration, made once.
		456	servers: Mutex<HashMap<String, Arc<ServerConfig>>>,
		457	}
		458
		459	fn provider() -> Arc<rustls::crypto::CryptoProvider> {
		460	Arc::new(rustls::crypto::ring::default_provider())
		461	}
		462
		463	impl Authority {
		464	/// A new authority that can sign for `hosts` alone, good for
		465	/// `valid_for` (the job's time, and a margin).
		466	pub(crate) fn new(hosts: &[String], valid_for: Duration) -> Result<Authority> {
		467	use rcgen::{BasicConstraints, CertificateParams, DistinguishedName, DnType, GeneralSubtree, IsCa, KeyPair, KeyUsagePurpose, NameConstraints};
		468	let now = time::OffsetDateTime::now_utc();
		469	let valid = (now - time::Duration::days(1), now + time::Duration::try_from(valid_for).unwrap_or(time::Duration::days(2)) + time::Duration::days(1));
		470	let key = KeyPair::generate().context("could not make a key")?;
		471	let mut params = CertificateParams::default();
		472	let mut name = DistinguishedName::new();
		473	name.push(DnType::CommonName, "g1t brokered secrets for one job");
		474	name.push(DnType::OrganizationName, "g1t");
		475	params.distinguished_name = name;
		476	params.is_ca = IsCa::Ca(BasicConstraints::Constrained(0));
		477	params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign, KeyUsagePurpose::DigitalSignature];
		478	(params.not_before, params.not_after) = valid;
		479	// It can sign for the brokered hosts and nothing else, so even its
		480	// key in other hands opens no other connection.
		481	let mut permitted: Vec<GeneralSubtree> = hosts
		482	.iter()
		483	.map(|host| host.trim().trim_start_matches("*.").to_ascii_lowercase())
		484	.filter(|host| !host.is_empty() && host.parse::<std::net::IpAddr>().is_err())
		485	.map(GeneralSubtree::DnsName)
		486	.collect();
		487	permitted.dedup();
		488	params.name_constraints = Some(NameConstraints { permitted_subtrees: permitted, excluded_subtrees: Vec::new() });
		489	let cert = params.self_signed(&key).context("could not make the job's certificate authority")?;
		490	Ok(Authority { cert, key, valid, servers: Mutex::new(HashMap::new()) })
		491	}
		492
		493	/// The authority's certificate, which the job's tools trust.
		494	pub(crate) fn pem(&self) -> String {
		495	self.cert.pem()
		496	}
		497
		498	#[cfg(test)]
		499	pub(crate) fn der(&self) -> CertificateDer<'static> {
		500	self.cert.der().clone()
		501	}
		502
		503	/// The TLS configuration that answers as `host`.
		504	fn server(&self, host: &str) -> Result<Arc<ServerConfig>> {
		505	let mut servers = self.servers.lock().map_err(|_| anyhow!("the certificates' lock is poisoned"))?;
		506	if let Some(config) = servers.get(host) {
		507	return Ok(config.clone());
		508	}
		509	use rcgen::{CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, KeyPair, KeyUsagePurpose};
		510	let key = KeyPair::generate()?;
		511	let mut params = CertificateParams::new(vec![host.to_owned()])?;
		512	let mut name = DistinguishedName::new();
		513	name.push(DnType::CommonName, host);
		514	params.distinguished_name = name;
		515	params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
		516	params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
		517	(params.not_before, params.not_after) = self.valid;
		518	let cert = params.signed_by(&key, &self.cert, &self.key)?;
		519	let private: PrivateKeyDer<'static> = PrivatePkcs8KeyDer::from(key.serialize_der()).into();
		520	let mut config = ServerConfig::builder_with_provider(provider())
		521	.with_safe_default_protocol_versions()?
		522	.with_no_client_auth()
		523	.with_single_cert(vec![cert.der().clone(), self.cert.der().clone()], private)?;
		524	// HTTP/1.1, which is all the proxy reads.
		525	config.alpn_protocols = vec![b"http/1.1".to_vec()];
		526	let config = Arc::new(config);
		527	servers.insert(host.to_owned(), config.clone());
		528	Ok(config)
		529	}
		530	}
		531
		532	/// How the proxy reaches a brokered host for a request without a stand-in:
		533	/// the system's trusted certificates, or the ones rustls ships without them.
		534	fn client_config(bundle: Option<&Path>) -> Result<Arc<ClientConfig>> {
		535	let mut roots = RootCertStore::empty();
		536	if let Some(bundle) = bundle
		537	&& let Ok(found) = CertificateDer::pem_file_iter(bundle)
		538	{
		539	roots.add_parsable_certificates(found.filter_map(std::result::Result::ok));
		540	}
		541	if roots.is_empty() {
		542	roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
		543	}
		544	let mut config = ClientConfig::builder_with_provider(provider()).with_safe_default_protocol_versions()?.with_root_certificates(roots).with_no_client_auth();
		545	config.alpn_protocols = vec![b"http/1.1".to_vec()];
		546	Ok(Arc::new(config))
		547	}
		548
		549	// --- The proxy -----------------------------------------------------------------
		550
		551	/// What every connection shares.
		552	struct Shared {
		553	stand_ins: Vec<StandIn>,
		554	relay: Relay,
		555	authority: Authority,
		556	client: Arc<ClientConfig>,
		557	open: AtomicUsize,
		558	}
		559
		560	impl Shared {
		561	fn brokered(&self, host: &str) -> bool {
		562	self.stand_ins.iter().any(|known| allows(&known.hosts, host))
		563	}
		564	}
		565
		566	/// A connection with bytes already read from it put back in front.
		567	struct Prefixed<S> {
		568	prefix: Vec<u8>,
		569	at: usize,
		570	inner: S,
		571	}
		572
		573	impl<S: Read> Read for Prefixed<S> {
		574	fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
		575	if self.at < self.prefix.len() {
		576	let n = buf.len().min(self.prefix.len() - self.at);
		577	buf[..n].copy_from_slice(&self.prefix[self.at..self.at + n]);
		578	self.at += n;
		579	return Ok(n);
		580	}
		581	self.inner.read(buf)
		582	}
		583	}
		584
		585	impl<S: Write> Write for Prefixed<S> {
		586	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
		587	self.inner.write(buf)
		588	}
		589	fn flush(&mut self) -> io::Result<()> {
		590	self.inner.flush()
		591	}
		592	}
		593
		594	/// `host:port` of a `CONNECT`, the port 443 when it names none.
		595	fn authority_of(target: &str) -> Option<(String, u16)> {
		596	let (host, port) = match target.rsplit_once(':') {
		597	Some((host, port)) if !port.contains(']') => (host, port.parse().ok()?),
		598	_ => (target, 443),
		599	};
		600	let host = host.trim_start_matches('[').trim_end_matches(']');
		601	(!host.is_empty()).then(|| (host.to_owned(), port))
		602	}
		603
		604	fn connect(host: &str, port: u16) -> io::Result<TcpStream> {
		605	let mut last = io::Error::new(io::ErrorKind::NotFound, format!("{host} has no address"));
		606	for address in (host, port).to_socket_addrs()? {
		607	match TcpStream::connect_timeout(&address, CONNECT_TIME) {
		608	Ok(stream) => return Ok(stream),
		609	Err(error) => last = error,
		610	}
		611	}
		612	Err(last)
		613	}
		614
		615	/// Copies both ways until either side is done.
		616	fn tunnel(client: TcpStream, prefix: Vec<u8>, upstream: TcpStream) -> io::Result<()> {
		617	let (mut client_in, mut upstream_out) = (client.try_clone()?, upstream.try_clone()?);
		618	upstream_out.write_all(&prefix)?;
		619	let up = std::thread::spawn(move || {
		620	let _ = io::copy(&mut client_in, &mut upstream_out);
		621	let _ = upstream_out.shutdown(Shutdown::Write);
		622	});
		623	let (mut upstream_in, mut client_out) = (upstream, client);
		624	let _ = io::copy(&mut upstream_in, &mut client_out);
		625	let _ = client_out.shutdown(Shutdown::Write);
		626	let _ = up.join();
		627	Ok(())
		628	}
		629
		630	/// One connection to the proxy.
		631	fn handle(stream: TcpStream, shared: &Shared) -> io::Result<()> {
		632	stream.set_read_timeout(Some(IDLE))?;
		633	stream.set_write_timeout(Some(IDLE))?;
		634	let mut reader = BufReader::new(stream.try_clone()?);
		635	let Some(head) = read_head(&mut reader)? else { return Ok(()) };
		636	let mut stream = stream;
		637	if !head.method.eq_ignore_ascii_case("CONNECT") {
		638	// Plain HTTP is never filled in, so it is not offered here: the
		639	// steps' HTTP_PROXY is left as the computer has it.
		640	return write_plain(&mut stream, 405, "this proxy takes HTTPS (CONNECT) only.");
		641	}
		642	let Some((host, port)) = authority_of(&head.target) else {
		643	return write_plain(&mut stream, 400, "the CONNECT target does not read.");
		644	};
		645	let prefix = reader.buffer().to_vec();
		646	if !shared.brokered(&host) {
		647	let upstream = match connect(&host, port) {
		648	Ok(upstream) => upstream,
		649	Err(error) => return write_plain(&mut stream, 502, &format!("{host} could not be reached: {error}")),
		650	};
		651	stream.write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")?;
		652	return tunnel(stream, prefix, upstream);
		653	}
		654	let config = match shared.authority.server(&normalize_host(&host)) {
		655	Ok(config) => config,
		656	Err(error) => return write_plain(&mut stream, 502, &format!("no certificate for {host}: {error}")),
		657	};
		658	stream.write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")?;
		659	let connection = ServerConnection::new(config).map_err(io::Error::other)?;
		660	let tls = StreamOwned::new(connection, Prefixed { prefix, at: 0, inner: stream });
		661	serve(tls, &host, port, shared)
		662	}
		663
		664	/// The requests on a brokered host's connection, one after another.
		665	fn serve<S: Read + Write>(tls: S, host: &str, port: u16, shared: &Shared) -> io::Result<()> {
		666	let mut reader = BufReader::new(tls);
		667	loop {
		668	let Some(head) = read_head(&mut reader)? else { return Ok(()) };
		669	if head.header("expect").is_some_and(|expect| expect.eq_ignore_ascii_case("100-continue")) {
		670	reader.get_mut().write_all(b"HTTP/1.1 100 Continue\r\n\r\n")?;
		671	reader.get_mut().flush()?;
		672	}
		673	let body = match read_body(&mut reader, &head) {
		674	Ok(body) => body,
		675	Err(error) if too_large(&error) => return write_plain(reader.get_mut(), 413, &error.to_string()),
		676	Err(error) => return Err(error),
		677	};
		678	if !carries(&head.target, &head.headers, &shared.stand_ins) {
		679	// Nothing of g1t's to fill in: straight to the host, and the
		680	// client opens a new connection for its next request.
		681	return direct(reader.get_mut(), &head, &body, host, port, &shared.client);
		682	}
		683	let url = url_of(host, port, &head.target);
		684	let headers: Vec<(String, String)> = head.headers.iter().filter(|(name, _)| !hop_by_hop(name) && !name.eq_ignore_ascii_case("host")).cloned().collect();
		685	let answer = shared.relay.send(&head.method, &url, &headers, &body);
		686	write_answer(reader.get_mut(), &head.method, answer)?;
		687	if head.closes() {
		688	return Ok(());
		689	}
		690	}
		691	}
		692
		693	/// The URL a request on a tunnel to `host:port` is for. A target written
		694	/// in full names its path only: the tunnel's host is the one the client
		695	/// asked for and the certificate was made for.
		696	fn url_of(host: &str, port: u16, target: &str) -> String {
		697	let path = match target.split_once("://") {
		698	Some((_, rest)) => rest.find('/').map_or("/", |at| &rest[at..]),
		699	None if target.starts_with('/') => target,
		700	None => "/",
		701	};
		702	let host = if host.contains(':') { format!("[{host}]") } else { host.to_owned() };
		703	if port == 443 { format!("https://{host}{path}") } else { format!("https://{host}:{port}{path}") }
		704	}
		705
		706	/// Sends a request without a stand-in to its host as it came, over a
		707	/// connection of its own that closes after the answer.
		708	fn direct(client: &mut impl Write, head: &Head, body: &[u8], host: &str, port: u16, config: &Arc<ClientConfig>) -> io::Result<()> {
		709	let name = match ServerName::try_from(host.to_owned()) {
		710	Ok(name) => name,
		711	Err(_) => return write_plain(client, 400, &format!("{host} is not a host name")),
		712	};
		713	let upstream = match connect(host, port) {
		714	Ok(upstream) => upstream,
		715	Err(error) => return write_plain(client, 502, &format!("{host} could not be reached: {error}")),
		716	};
		717	upstream.set_read_timeout(Some(IDLE))?;
		718	let connection = ClientConnection::new(config.clone(), name).map_err(io::Error::other)?;
		719	let mut tls = StreamOwned::new(connection, upstream);
		720	let mut request = format!("{} {} HTTP/1.1\r\n", head.method, head.target);
		721	for (name, value) in head.headers.iter().filter(|(name, _)| !hop_by_hop(name)) {
		722	request.push_str(&format!("{name}: {value}\r\n"));
		723	}
		724	if !body.is_empty() || !matches!(head.method.to_ascii_uppercase().as_str(), "GET" | "HEAD" | "OPTIONS" | "DELETE") {
		725	request.push_str(&format!("content-length: {}\r\n", body.len()));
		726	}
		727	request.push_str("connection: close\r\n\r\n");
		728	tls.write_all(request.as_bytes())?;
		729	tls.write_all(body)?;
		730	tls.flush()?;
		731	match io::copy(&mut tls, client) {
		732	// A host that closes without saying so has still answered.
		733	Err(error) if error.kind() != io::ErrorKind::UnexpectedEof => return Err(error),
		734	_ => {}
		735	}
		736	client.flush()
		737	}
		738
		739	/// The proxy, running.
		740	pub(crate) struct Proxy {
		741	pub(crate) port: u16,
		742	}
		743
		744	fn listen(shared: Shared) -> Result<Proxy> {
		745	let listener = TcpListener::bind(("127.0.0.1", 0)).context("could not listen on 127.0.0.1")?;
		746	let port = listener.local_addr()?.port();
		747	let shared = Arc::new(shared);
		748	std::thread::spawn(move || {
		749	for stream in listener.incoming().flatten() {
		750	if shared.open.load(Ordering::SeqCst) >= MAX_CONNECTIONS {
		751	continue;
		752	}
		753	shared.open.fetch_add(1, Ordering::SeqCst);
		754	let shared = shared.clone();
		755	std::thread::spawn(move || {
		756	let _ = handle(stream, &shared);
		757	shared.open.fetch_sub(1, Ordering::SeqCst);
		758	});
		759	}
		760	});
		761	Ok(Proxy { port })
		762	}
		763
		764	/// The proxy for `stand_ins`, and the variables the job's steps get:
		765	/// `HTTPS_PROXY` and the trusted-certificate files, written to `folder`.
		766	/// `env` is the steps' own environment, whose certificate files are kept
		767	/// in what they are pointed at.
		768	pub(crate) fn start(stand_ins: Vec<StandIn>, relay: Relay, folder: &Path, env: &BTreeMap<String, String>, valid_for: Duration) -> Result<(Proxy, Vec<(String, String)>)> {
		769	let hosts: Vec<String> = stand_ins.iter().flat_map(|known| known.hosts.iter().cloned()).collect();
		770	let authority = Authority::new(&hosts, valid_for)?;
		771	std::fs::create_dir_all(folder).context("could not make the certificates' folder")?;
		772	let pem = authority.pem();
		773	let system = system_bundle(env);
		774	let client = client_config(system.as_deref())?;
		775	let proxy = listen(Shared { stand_ins, relay, authority, client, open: AtomicUsize::new(0) })?;
		776	let mut vars = trust_variables(folder, &pem, env, system.as_deref())?;
		777	let address = format!("http://127.0.0.1:{}", proxy.port);
		778	vars.push(("HTTPS_PROXY".into(), address.clone()));
		779	vars.push(("https_proxy".into(), address));
		780	// Node's own fetch and https take a proxy from these only when asked.
		781	vars.push(("NODE_USE_ENV_PROXY".into(), "1".into()));
		782	// Loopback never goes through it, unless the steps already say what doesn't.
		783	if !env.contains_key("NO_PROXY") && !env.contains_key("no_proxy") {
		784	vars.push(("NO_PROXY".into(), "localhost,127.0.0.1,::1".into()));
		785	vars.push(("no_proxy".into(), "localhost,127.0.0.1,::1".into()));
		786	}
		787	Ok((proxy, vars))
		788	}
		789
		790	/// The bundle of certificates the steps trust today: `SSL_CERT_FILE`, or
		791	/// the system's.
		792	fn system_bundle(env: &BTreeMap<String, String>) -> Option<PathBuf> {
		793	env.get("SSL_CERT_FILE").map(PathBuf::from).into_iter().chain(SYSTEM_BUNDLES.iter().map(PathBuf::from)).find(|path| path.is_file())
		794	}
		795
		796	/// The trusted-certificate variables: each pointed at what it pointed at
		797	/// before (or the system's bundle) with the job's authority added, written
		798	/// once per source. Without a bundle to add to, only `NODE_EXTRA_CA_CERTS`,
		799	/// which adds to Node's own.
		800	fn trust_variables(folder: &Path, pem: &str, env: &BTreeMap<String, String>, system: Option<&Path>) -> Result<Vec<(String, String)>> {
		801	let mut written: HashMap<PathBuf, PathBuf> = HashMap::new();
		802	let mut vars = Vec::new();
		803	let mut add_to = |source: &Path, name: &str| -> Result<String> {
		804	if let Some(out) = written.get(source) {
		805	return Ok(out.display().to_string());
		806	}
		807	let mut combined = std::fs::read(source).with_context(|| format!("could not read {}", source.display()))?;
		808	if !combined.ends_with(b"\n") && !combined.is_empty() {
		809	combined.push(b'\n');
		810	}
		811	combined.extend_from_slice(pem.as_bytes());
		812	let out = folder.join(format!("{}-{}.pem", written.len() + 1, name.to_ascii_lowercase()));
		813	std::fs::write(&out, combined)?;
		814	written.insert(source.to_owned(), out.clone());
		815	Ok(out.display().to_string())
		816	};
		817	for name in BUNDLE_VARIABLES {
		818	let own = env.get(*name).map(PathBuf::from).filter(|path| path.is_file());
		819	if let Some(source) = own.as_deref().or(system) {
		820	vars.push(((*name).to_owned(), add_to(source, name)?));
		821	}
		822	}
		823	let authority = folder.join("g1t-broker-ca.pem");
		824	std::fs::write(&authority, pem)?;
		825	let node = match env.get("NODE_EXTRA_CA_CERTS").map(PathBuf::from).filter(|path| path.is_file()) {
		826	Some(own) => add_to(&own, "NODE_EXTRA_CA_CERTS")?,
		827	None => authority.display().to_string(),
		828	};
		829	vars.push(("NODE_EXTRA_CA_CERTS".into(), node));
		830	Ok(vars)
		831	}
		832
		833	/// The names of the variables `start` sets, which a container a step
		834	/// starts (a Docker action, a service) does not get: the proxy is on this
		835	/// harness's own loopback, out of their reach.
		836	pub(crate) fn names(vars: &[(String, String)]) -> impl Iterator<Item = String> + '_ {
		837	vars.iter().map(|(name, _)| name.clone())
		838	}
		839
		840	#[cfg(test)]
		841	mod tests {
		842	use std::io::{BufRead, BufReader, Read, Write};
		843	use std::net::{TcpListener, TcpStream};
		844	use std::sync::Arc;
		845	use std::sync::mpsc;
		846
		847	use rustls::pki_types::ServerName;
		848	use rustls::{ClientConfig, ClientConnection, RootCertStore, StreamOwned};
		849	use serde_json::{Value, json};
		850
		851	use super::*;
		852
		853	const STAND_IN: &str = "g1tsec_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
		854
		855	fn npm() -> StandIn {
		856	StandIn { stand_in: STAND_IN.into(), name: "NPM_TOKEN".into(), hosts: vec!["registry.npmjs.org".into(), "*.example.com".into()] }
		857	}
		858
		859	#[test]
		860	fn the_spec_gives_stand_ins_and_hosts() {
		861	let spec = json!({ "brokered": [
		862	{ "stand_in": STAND_IN, "name": "NPM_TOKEN", "hosts": ["registry.npmjs.org", "*.example.com"] },
		863	{ "stand_in": "not-a-stand-in", "name": "X", "hosts": ["x.example.com"] },
		864	{ "stand_in": "g1tsec_b", "name": "NO_HOSTS", "hosts": [] },
		865	] });
		866	assert_eq!(stand_ins(&spec), vec![npm()]);
		867	assert!(stand_ins(&json!({})).is_empty());
		868	}
		869
		870	#[test]
		871	fn hosts_match_as_g1t_reads_them() {
		872	let hosts = npm().hosts;
		873	assert!(allows(&hosts, "registry.npmjs.org"));
		874	assert!(allows(&hosts, "Registry.NPMJS.org.:443"));
		875	assert!(allows(&hosts, "a.b.example.com"));
		876	assert!(!allows(&hosts, "example.com"), "a wildcard is not its own apex");
		877	assert!(!allows(&hosts, "evil-example.com"));
		878	assert!(!allows(&hosts, "registry.npmjs.org.evil.com"));
		879	assert_eq!(normalize_host("[::1]:443"), "[::1]");
		880	assert_eq!(authority_of("registry.npmjs.org:443"), Some(("registry.npmjs.org".into(), 443)));
		881	assert_eq!(authority_of("[::1]:8443"), Some(("::1".into(), 8443)));
		882	assert_eq!(authority_of("example.com"), Some(("example.com".into(), 443)));
		883	assert_eq!(url_of("registry.npmjs.org", 443, "/pkg?x=1"), "https://registry.npmjs.org/pkg?x=1");
		884	assert_eq!(url_of("localhost", 8443, "/"), "https://localhost:8443/");
		885	assert_eq!(url_of("::1", 443, "/a"), "https://[::1]/a");
		886	// A full URL inside the tunnel cannot send the request elsewhere.
		887	assert_eq!(url_of("registry.npmjs.org", 443, "https://evil.example/steal?t=1"), "https://registry.npmjs.org/steal?t=1");
		888	}
		889
		890	#[test]
		891	fn stand_ins_are_found_in_the_target_headers_and_basic_credentials() {
		892	let known = [npm()];
		893	let header = |name: &str, value: &str| vec![(name.to_owned(), value.to_owned())];
		894	assert!(carries(&format!("/pkg?token={STAND_IN}"), &[], &known));
		895	assert!(carries("/pkg", &header("authorization", &format!("Bearer {STAND_IN}")), &known));
		896	let basic = format!("Basic {}", BASE64.encode(format!("me:{STAND_IN}")));
		897	assert!(carries("/pkg", &header("authorization", &basic), &known));
		898	assert!(!carries("/pkg", &header("authorization", "Bearer g1tsec_other"), &known));
		899	}
		900
		901	#[test]
		902	fn only_unix_without_a_proxy_of_its_own_brokers() {
		903	assert_eq!(possible(|_| None), cfg!(unix));
		904	assert!(!possible(|name| (name == "HTTPS_PROXY").then(|| "http://corp:3128".into())));
		905	assert_eq!(possible(|name| (name == "https_proxy").then(String::new)), cfg!(unix));
		906	}
		907
		908	#[test]
		909	fn heads_and_bodies_read_by_length_or_in_chunks_within_limits() {
		910	let mut input = BufReader::new(&b"\r\nPOST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n3\r\nabc\r\n2;x=y\r\nde\r\n0\r\nT: 1\r\n\r\nGET /b HTTP/1.1\r\nContent-Length: 2\r\n\r\nhi"[..]);
		911	let head = read_head(&mut input).unwrap().unwrap();
		912	assert_eq!((head.method.as_str(), head.target.as_str()), ("POST", "/a"));
		913	assert_eq!(read_body(&mut input, &head).unwrap(), b"abcde");
		914	let next = read_head(&mut input).unwrap().unwrap();
		915	assert_eq!(read_body(&mut input, &next).unwrap(), b"hi");
		916	assert!(read_head(&mut input).unwrap().is_none());
		917
		918	let big = format!("PUT / HTTP/1.1\r\nContent-Length: {}\r\n\r\n", MAX_BODY_BYTES + 1);
		919	let mut input = BufReader::new(big.as_bytes());
		920	let head = read_head(&mut input).unwrap().unwrap();
		921	assert!(too_large(&read_body(&mut input, &head).unwrap_err()));
		922	let long = format!("GET / HTTP/1.1\r\nX: {}\r\n\r\n", "a".repeat(MAX_HEAD_BYTES));
		923	assert!(read_head(&mut BufReader::new(long.as_bytes())).is_err());
		924	assert!(Head { method: "GET".into(), target: "/".into(), version: "HTTP/1.0".into(), headers: vec![] }.closes());
		925	}
		926
		927	#[test]
		928	fn the_authority_signs_only_for_the_brokered_hosts_and_the_leaf_is_trusted() {
		929	let authority = Authority::new(&["localhost".into(), "*.example.com".into()], Duration::from_secs(3600)).unwrap();
		930	assert!(authority.pem().starts_with("-----BEGIN CERTIFICATE-----"));
		931	let config = authority.server("localhost").unwrap();
		932	assert!(Arc::ptr_eq(&config, &authority.server("localhost").unwrap()), "made once per host");
		933
		934	// A handshake as `host`, with a client that trusts the authority.
		935	let handshake = |host: &str| -> Result<(), String> {
		936	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
		937	let port = listener.local_addr().unwrap().port();
		938	let server = authority.server(host).unwrap();
		939	let serving = std::thread::spawn(move || {
		940	let stream = listener.incoming().flatten().next().unwrap();
		941	let mut tls = StreamOwned::new(rustls::ServerConnection::new(server).unwrap(), stream);
		942	let _ = tls.write_all(b"x");
		943	});
		944	let mut roots = RootCertStore::empty();
		945	roots.add(authority.der()).unwrap();
		946	let config = ClientConfig::builder_with_provider(provider()).with_safe_default_protocol_versions().unwrap().with_root_certificates(roots).with_no_client_auth();
		947	let stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
		948	let mut tls = StreamOwned::new(ClientConnection::new(Arc::new(config), ServerName::try_from(host.to_owned()).unwrap()).unwrap(), stream);
		949	let mut byte = [0; 1];
		950	let read = tls.read_exact(&mut byte).map_err(|error| error.to_string());
		951	let _ = serving.join();
		952	read
		953	};
		954	assert_eq!(handshake("localhost"), Ok(()));
		955	assert_eq!(handshake("a.example.com"), Ok(()));
		956	// Its key can sign for another host, but nothing that trusts it accepts that.
		957	let refused = handshake("bank.test").unwrap_err();
		958	assert!(refused.to_lowercase().contains("name"), "{refused}");
		959	}
		960
		961	/// A stand-in for g1t's broker: answers each request with the body it
		962	/// was given, its status and headers in `x-g1t-broker-response`, and
		963	/// tells the test what it was asked.
		964	/// What the fake g1t heard: a request's headers and body.
		965	type Heard = mpsc::Receiver<(Vec<(String, String)>, Vec<u8>)>;
		966
		967	fn fake_g1t(answer_status: u16) -> (String, Heard) {
		968	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
		969	let base = format!("http://{}", listener.local_addr().unwrap());
		970	let (tell, heard) = mpsc::channel();
		971	std::thread::spawn(move || {
		972	for stream in listener.incoming().flatten() {
		973	let mut reader = BufReader::new(stream.try_clone().unwrap());
		974	let head = read_head(&mut reader).unwrap().unwrap();
		975	let body = read_body(&mut reader, &head).unwrap();
		976	tell.send((head.headers.clone(), body.clone())).unwrap();
		977	let answer = BASE64.encode(json!({ "status": answer_status, "headers": [["content-type", "text/plain"], ["x-from", "upstream"]] }).to_string());
		978	let mut stream = stream;
		979	let reply = format!("HTTP/1.1 200 OK\r\n{ANSWER_HEADER}: {answer}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", body.len() + 4);
		980	stream.write_all(reply.as_bytes()).unwrap();
		981	stream.write_all(b"got ").unwrap();
		982	stream.write_all(&body).unwrap();
		983	}
		984	});
		985	(base, heard)
		986	}
		987
		988	fn proxy_for(base: &str, stand_ins: Vec<StandIn>, client: Arc<ClientConfig>) -> (Proxy, Arc<ClientConfig>) {
		989	let api = Api { base: base.into(), job: "job_1".into(), token: "current-token".into() };
		990	let hosts: Vec<String> = stand_ins.iter().flat_map(|s| s.hosts.clone()).collect();
		991	let authority = Authority::new(&hosts, Duration::from_secs(3600)).unwrap();
		992	let mut roots = RootCertStore::empty();
		993	roots.add(authority.der()).unwrap();
		994	let trusting = ClientConfig::builder_with_provider(provider()).with_safe_default_protocol_versions().unwrap().with_root_certificates(roots).with_no_client_auth();
		995	let proxy = listen(Shared { stand_ins, relay: Relay::new(&api), authority, client, open: AtomicUsize::new(0) }).unwrap();
		996	(proxy, Arc::new(trusting))
		997	}
		998
		999	/// Opens `host:port` through the proxy with CONNECT.
		1000	fn through(proxy: &Proxy, target: &str) -> TcpStream {
		1001	let mut stream = TcpStream::connect(("127.0.0.1", proxy.port)).unwrap();
		1002	write!(stream, "CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n").unwrap();
		1003	let mut reader = BufReader::new(stream.try_clone().unwrap());
		1004	let mut line = String::new();
		1005	reader.read_line(&mut line).unwrap();
		1006	assert!(line.starts_with("HTTP/1.1 200"), "{line}");
		1007	reader.read_line(&mut line).unwrap();
		1008	stream
		1009	}
		1010
		1011	fn read_all(mut reader: impl Read) -> String {
		1012	let mut text = Vec::new();
		1013	let _ = reader.read_to_end(&mut text);
		1014	String::from_utf8_lossy(&text).into_owned()
		1015	}
		1016
		1017	#[test]
		1018	fn a_request_with_a_stand_in_goes_to_g1t_and_its_answer_comes_back() {
		1019	let (base, heard) = fake_g1t(201);
		1020	let known = StandIn { stand_in: STAND_IN.into(), name: "NPM_TOKEN".into(), hosts: vec!["localhost".into()] };
		1021	let (proxy, trusting) = proxy_for(&base, vec![known], client_config(None).unwrap());
		1022	let stream = through(&proxy, "localhost:8443");
		1023	let connection = ClientConnection::new(trusting, ServerName::try_from("localhost").unwrap()).unwrap();
		1024	let mut tls = StreamOwned::new(connection, stream);
		1025	write!(tls, "POST /publish?x=1 HTTP/1.1\r\nHost: localhost:8443\r\nAuthorization: Bearer {STAND_IN}\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello").unwrap();
		1026	let answer = read_all(&mut tls);
		1027	assert!(answer.starts_with("HTTP/1.1 201 Created\r\n"), "{answer}");
		1028	assert!(answer.contains("x-from: upstream\r\n") && answer.contains("transfer-encoding: chunked"), "{answer}");
The runner's broker test reads the relayed answer's body as chunks: the proxy passes the body on as it reads it, so "got " and "hello" can arrive as two chunks, and looking for "got hello" in the raw bytes failed CI #166 on a correct answer.12h ago		1029	// The body may come in more than one chunk, so it is read as chunks.
		1030	let mut chunks = BufReader::new(answer.as_bytes());
		1031	let head = read_head(&mut chunks).unwrap().unwrap();
		1032	assert_eq!(read_body(&mut chunks, &head).unwrap(), b"got hello", "{answer}");
Merge branch 'main' into rm-agents12h ago		1033
		1034	let (headers, body) = heard.recv().unwrap();
		1035	assert_eq!(body, b"hello");
		1036	let header = |name: &str| headers.iter().find(|(n, _)| n.eq_ignore_ascii_case(name)).map(|(_, v)| v.clone());
		1037	assert_eq!(header("authorization").as_deref(), Some("Bearer current-token"), "the job's token in a header");
		1038	let wanted: Value = serde_json::from_slice(&BASE64.decode(header(REQUEST_HEADER).unwrap()).unwrap()).unwrap();
		1039	assert_eq!(wanted["method"], "POST");
		1040	assert_eq!(wanted["url"], "https://localhost:8443/publish?x=1");
		1041	let sent: Vec<(String, String)> = serde_json::from_value(wanted["headers"].clone()).unwrap();
		1042	assert!(sent.contains(&("Authorization".into(), format!("Bearer {STAND_IN}"))), "the stand-in, never a value: {sent:?}");
		1043	assert!(!sent.iter().any(|(n, _)| n.eq_ignore_ascii_case("host") || n.eq_ignore_ascii_case("connection") || n.eq_ignore_ascii_case("content-length")));
		1044	}
		1045
		1046	#[test]
		1047	fn g1t_s_refusal_reaches_the_client_as_it_said_it() {
		1048	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
		1049	let base = format!("http://{}", listener.local_addr().unwrap());
		1050	std::thread::spawn(move || {
		1051	for mut stream in listener.incoming().flatten() {
		1052	let mut reader = BufReader::new(stream.try_clone().unwrap());
		1053	let head = read_head(&mut reader).unwrap().unwrap();
		1054	read_body(&mut reader, &head).unwrap();
		1055	let body = r#"{"error":{"code":"forbidden","message":"not for this host"}}"#;
		1056	write!(stream, "HTTP/1.1 403 Forbidden\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).unwrap();
		1057	}
		1058	});
		1059	let api = Api { base, job: "job_1".into(), token: "t".into() };
		1060	match Relay::new(&api).send("GET", "https://localhost/", &[], b"") {
		1061	Answer::Refused { status, message } => assert_eq!((status, message.as_str()), (403, "not for this host")),
		1062	Answer::Relayed { .. } => panic!("refused"),
		1063	}
		1064	// g1t out of reach: a bad gateway, saying so.
		1065	let api = Api { base: "http://127.0.0.1:1".into(), job: "job_1".into(), token: "t".into() };
		1066	assert!(matches!(Relay::new(&api).send("GET", "https://localhost/", &[], b""), Answer::Refused { status: 502, .. }));
		1067	}
		1068
		1069	#[test]
		1070	fn other_hosts_are_tunnelled_untouched_and_plain_http_is_refused() {
		1071	let echo = TcpListener::bind("127.0.0.1:0").unwrap();
		1072	let port = echo.local_addr().unwrap().port();
		1073	std::thread::spawn(move || {
		1074	for mut stream in echo.incoming().flatten() {
		1075	let mut buffer = [0; 5];
		1076	stream.read_exact(&mut buffer).unwrap();
		1077	stream.write_all(&buffer).unwrap();
		1078	}
		1079	});
		1080	let (proxy, _) = proxy_for("http://127.0.0.1:1", vec![npm()], client_config(None).unwrap());
		1081	let mut stream = through(&proxy, &format!("127.0.0.1:{port}"));
		1082	stream.write_all(b"bytes").unwrap();
		1083	let mut back = [0; 5];
		1084	stream.read_exact(&mut back).unwrap();
		1085	assert_eq!(&back, b"bytes");
		1086
		1087	let mut plain = TcpStream::connect(("127.0.0.1", proxy.port)).unwrap();
		1088	plain.write_all(b"GET http://registry.npmjs.org/ HTTP/1.1\r\nHost: registry.npmjs.org\r\n\r\n").unwrap();
		1089	assert!(read_all(plain).starts_with("HTTP/1.1 405"));
		1090	}
		1091
		1092	#[test]
		1093	fn a_request_without_a_stand_in_goes_straight_to_the_host() {
		1094	// The host: TLS with a certificate of its own authority.
		1095	let host_ca = Authority::new(&["localhost".into()], Duration::from_secs(3600)).unwrap();
		1096	let host_config = host_ca.server("localhost").unwrap();
		1097	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
		1098	let port = listener.local_addr().unwrap().port();
		1099	let (tell, heard) = mpsc::channel();
		1100	std::thread::spawn(move || {
		1101	let stream = listener.incoming().flatten().next().unwrap();
		1102	let connection = rustls::ServerConnection::new(host_config).unwrap();
		1103	let mut tls = BufReader::new(StreamOwned::new(connection, stream));
		1104	let head = read_head(&mut tls).unwrap().unwrap();
		1105	tell.send(head).unwrap();
		1106	tls.get_mut().write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 6\r\nconnection: close\r\n\r\ndirect").unwrap();
		1107	tls.get_mut().conn.send_close_notify();
		1108	let _ = tls.get_mut().flush();
		1109	});
		1110	let mut roots = RootCertStore::empty();
		1111	roots.add(host_ca.der()).unwrap();
		1112	let upstream = Arc::new(ClientConfig::builder_with_provider(provider()).with_safe_default_protocol_versions().unwrap().with_root_certificates(roots).with_no_client_auth());
		1113	let known = StandIn { stand_in: STAND_IN.into(), name: "NPM_TOKEN".into(), hosts: vec!["localhost".into()] };
		1114	let (proxy, trusting) = proxy_for("http://127.0.0.1:1", vec![known], upstream);
		1115	let stream = through(&proxy, &format!("localhost:{port}"));
		1116	let connection = ClientConnection::new(trusting, ServerName::try_from("localhost").unwrap()).unwrap();
		1117	let mut tls = StreamOwned::new(connection, stream);
		1118	write!(tls, "GET /public HTTP/1.1\r\nHost: localhost\r\nAccept: */*\r\n\r\n").unwrap();
		1119	let answer = read_all(&mut tls);
		1120	assert!(answer.starts_with("HTTP/1.1 200 OK") && answer.ends_with("direct"), "{answer}");
		1121	let head = heard.recv().unwrap();
		1122	assert_eq!(head.target, "/public");
		1123	assert_eq!(head.header("connection"), Some("close"));
		1124	}
		1125
		1126	#[test]
		1127	fn the_steps_trust_the_authority_beside_what_they_trusted_before() {
		1128	let folder = std::env::temp_dir().join(format!("g1t-broker-test-{}", std::process::id()));
		1129	std::fs::create_dir_all(&folder).unwrap();
		1130	let system = folder.join("system.pem");
		1131	std::fs::write(&system, "SYSTEM\n").unwrap();
		1132	let own = folder.join("own.pem");
		1133	std::fs::write(&own, "OWN").unwrap();
		1134	let env: BTreeMap<String, String> = [("REQUESTS_CA_BUNDLE".to_owned(), own.display().to_string())].into_iter().collect();
		1135	let vars: BTreeMap<String, String> = trust_variables(&folder, "AUTHORITY\n", &env, Some(&system)).unwrap().into_iter().collect();
		1136	assert_eq!(std::fs::read_to_string(&vars["SSL_CERT_FILE"]).unwrap(), "SYSTEM\nAUTHORITY\n");
		1137	assert_eq!(vars["SSL_CERT_FILE"], vars["GIT_SSL_CAINFO"], "one file per source");
		1138	assert_eq!(std::fs::read_to_string(&vars["REQUESTS_CA_BUNDLE"]).unwrap(), "OWN\nAUTHORITY\n");
		1139	assert_eq!(std::fs::read_to_string(&vars["NODE_EXTRA_CA_CERTS"]).unwrap(), "AUTHORITY\n");
		1140	// Without a bundle, Node's alone.
		1141	let vars = trust_variables(&folder, "AUTHORITY\n", &BTreeMap::new(), None).unwrap();
		1142	assert_eq!(vars.iter().map(|(n, _)| n.as_str()).collect::<Vec<_>>(), ["NODE_EXTRA_CA_CERTS"]);
		1143	let _ = std::fs::remove_dir_all(&folder);
		1144	}
		1145	}

This file's history is long; its oldest lines are credited to the oldest commit read.