flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh

git agents cloudflare rust forge mcp

Commit

Brokered secrets keep working after a job's token is swapped. The runner Worker looks up a job's stand-ins (brokered_secrets) with the token the job started with, which a harness that asks for rotation spends on its first spec fetch. The swap now keeps that token's hash beside the new one (first_token_hash, migration 0020), and brokered_secrets alone accepts it; reporting, logs, artifacts, cache, job_auth, actions and the spec itself take only the current token, and no API route passes brokered_secrets on, so the kept token is reachable only through g1t's own service bindings. It is cleared when the job ends, is cancelled, is claimed again or is re-run. The self-hosted runners guide says which lookup still takes it.

syntaqxcommitted 2d agoParent47627cfBrowse files

31	31		const MAX_CHUNK_BYTES: usize = 256 * 1024;
32	32		const MAX_ANNOTATIONS: usize = 50;
33	33		/// Swaps a running job's token for a new one, only while it still has the
34		−	/// one the caller showed (`job_spec` with `rotate`).
35		−	const TOKEN_ROTATE: &str = "UPDATE jobs SET token_hash = ?1 WHERE id = ?2 AND status = 'in_progress' AND token_hash = ?3 RETURNING id";
	34	+	/// one the caller showed (`job_spec` with `rotate`), keeping the hash of
	35	+	/// the one it started with for `brokered_secrets` (migration 0020).
	36	+	const TOKEN_ROTATE: &str = "UPDATE jobs SET first_token_hash = COALESCE(first_token_hash, token_hash), token_hash = ?1
	37	+	WHERE id = ?2 AND status = 'in_progress' AND token_hash = ?3 RETURNING id";
	38	+
	39	+	/// Whether `token` is a running job's: its current one, or with `started`
	40	+	/// (`brokered_secrets` only) the one it started with before its harness
	41	+	/// swapped it.
	42	+	fn token_opens(current: Option<&str>, first: Option<&str>, token: &str, started: bool) -> bool {
	43	+	let hash = sha256_hex(token);
	44	+	current.is_some_and(\|h\| same(h, &hash)) \|\| (started && first.is_some_and(\|h\| same(h, &hash)))
	45	+	}
36	46		/// A batch of a job's log, under the next `seq` of that job's, worked out
37	47		/// in the same statement (SQLite runs one write at a time). `?4` is the
38	48		/// runner's number for the batch (null from runners that send none): a
⋯
239	249		/// 0019), so its log is masked for their values too.
240	250		#[serde(default)]
241	251		pub brokered: u32,
	252	+	/// The hash of the token it started with, once its harness swapped that
	253	+	/// for another (migration 0020): good for `brokered_secrets` only.
	254	+	#[serde(default)]
	255	+	pub first_token_hash: Option<String>,
242	256		}
243	257
244	258		/// How long a cancelled job has to run its `if: always()` and `cancelled()`
⋯
1611	1625		let claimed = self
1612	1626		.db
1613	1627		.prepare(
1614		−	"UPDATE jobs SET status = 'in_progress', token_hash = ?, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
	1628	+	"UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
1615	1629		)
1616	1630		.bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1617	1631		.first::<Value>(None)
⋯
1678	1692		.db
1679	1693		.prepare(
1680	1694		"UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
1681		−	finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
	1695	+	finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
1682	1696		)
1683	1697		.bind(&[
1684	1698		conclusion.into(),
⋯
1786	1800		}
1787	1801		let stopped = self
1788	1802		.db
1789		−	.prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")
	1803	+	.prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")
1790	1804		.bind(&[reason.into(), now().into(), job.id.as_str().into()])?
1791	1805		.first::<JobRow>(None)
1792	1806		.await?;
⋯
2219	2233		self.db
2220	2234		.prepare(
2221	2235		"UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
2222		−	matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
	2236	+	matrix = NULL, call = NULL, token_hash = NULL, first_token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
2223	2237		labels = NULL, queued_at = NULL, runner_id = NULL, runner_name = NULL, environment = NULL,
2224	2238		concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
2225	2239		)
⋯
2260	2274		// --- The sandbox's side -----------------------------------------------------
2261	2275
2262	2276		pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
	2277	+	self.job_for(a, false).await
	2278	+	}
	2279	+
	2280	+	/// As [`Self::job_for_token`], and also with the token the job started
	2281	+	/// with after its harness swapped it (`job_spec` with `rotate`): for
	2282	+	/// `brokered_secrets` alone, which the runner Worker asks with the
	2283	+	/// token it was given at the start, and which no API route passes on.
	2284	+	pub(crate) async fn job_for_started_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
	2285	+	self.job_for(a, true).await
	2286	+	}
	2287	+
	2288	+	async fn job_for(&self, a: &JobCallArgs, started: bool) -> Result<Outcome<JobRow>> {
2263	2289		let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
2264	2290		Ok(match job {
2265		−	Some(job) if job.status == "in_progress" && job.token_hash.as_deref().is_some_and(\|hash\| same(hash, &sha256_hex(&a.token))) => {
2266		−	Outcome::Ok(job)
2267		−	}
	2291	+	Some(job) if job.status == "in_progress" && token_opens(job.token_hash.as_deref(), job.first_token_hash.as_deref(), &a.token, started) => Outcome::Ok(job),
2268	2292		_ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
2269	2293		})
2270	2294		}
⋯
2876	2900		use g1t_secrets::sha256_hex;
2877	2901		use rusqlite::{Connection, OptionalExtension, params};
2878	2902
2879		−	use super::TOKEN_ROTATE;
	2903	+	use super::{TOKEN_ROTATE, token_opens};
2880	2904
2881	2905		fn database() -> Connection {
2882	2906		let db = Connection::open_in_memory().unwrap();
2883		−	let path = format!("{}/migrations/0001_init.sql", env!("CARGO_MANIFEST_DIR"));
2884		−	db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
	2907	+	for name in ["0001_init.sql", "0020_job_token_rotation.sql"] {
	2908	+	let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
	2909	+	db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
	2910	+	}
2885	2911		db.execute(
2886	2912		"INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, status, token_hash) VALUES ('job_1', 'run_1', 'repo_1', 'acme', 'build', 'build', 'in_progress', ?1)",
2887	2913		[sha256_hex("first")],
⋯
2897	2923		.is_some()
2898	2924		}
2899	2925
2900		−	fn hash(db: &Connection) -> Option<String> {
2901		−	db.query_row("SELECT token_hash FROM jobs WHERE id = 'job_1'", [], \|row\| row.get(0)).unwrap()
	2926	+	/// The job's current token's hash, and the one it started with.
	2927	+	fn hashes(db: &Connection) -> (Option<String>, Option<String>) {
	2928	+	db.query_row("SELECT token_hash, first_token_hash FROM jobs WHERE id = 'job_1'", [], \|row\| Ok((row.get(0)?, row.get(1)?))).unwrap()
2902	2929		}
2903	2930
2904	2931		#[test]
2905	2932		fn a_job_token_fetches_its_spec_once() {
2906	2933		let db = database();
2907	2934		assert!(rotate(&db, "second", "first"));
2908		−	assert_eq!(hash(&db), Some(sha256_hex("second")));
	2935	+	assert_eq!(hashes(&db), (Some(sha256_hex("second")), Some(sha256_hex("first"))));
2909	2936		// The first token, shown again (a copy left on the computer, or a
2910	2937		// second fetch racing the first), swaps nothing and is refused.
2911	2938		assert!(!rotate(&db, "third", "first"));
2912		−	assert_eq!(hash(&db), Some(sha256_hex("second")));
	2939	+	assert_eq!(hashes(&db).0, Some(sha256_hex("second")));
2913	2940		// A job that ended has no token to swap.
2914		−	db.execute("UPDATE jobs SET status = 'completed', token_hash = NULL", []).unwrap();
	2941	+	db.execute("UPDATE jobs SET status = 'completed', token_hash = NULL, first_token_hash = NULL", []).unwrap();
2915	2942		assert!(!rotate(&db, "fourth", "second"));
2916		−	assert_eq!(hash(&db), None);
	2943	+	assert_eq!(hashes(&db), (None, None));
	2944	+	}
	2945	+
	2946	+	#[test]
	2947	+	fn the_worker_finds_brokered_secrets_with_the_token_the_job_started_with() {
	2948	+	let db = database();
	2949	+	let opens = \|db: &Connection, token: &str, started: bool\| {
	2950	+	let (current, first) = hashes(db);
	2951	+	token_opens(current.as_deref(), first.as_deref(), token, started)
	2952	+	};
	2953	+	// Before any swap, the job's one token opens everything.
	2954	+	assert!(opens(&db, "first", false) && opens(&db, "first", true));
	2955	+	assert!(rotate(&db, "second", "first"));
	2956	+	// The harness's new token opens everything.
	2957	+	assert!(opens(&db, "second", false) && opens(&db, "second", true));
	2958	+	// The token it started with, which the runner Worker holds, still
	2959	+	// finds the job's brokered secrets, and opens nothing else.
	2960	+	assert!(opens(&db, "first", true));
	2961	+	assert!(!opens(&db, "first", false));
	2962	+	assert!(!opens(&db, "other", true));
	2963	+	// Another swap keeps the token it started with, not the one between.
	2964	+	assert!(rotate(&db, "third", "second"));
	2965	+	assert!(opens(&db, "first", true));
	2966	+	assert!(!opens(&db, "second", true));
2917	2967		}
2918	2968
2919	2969		#[test]

342	342		folder Windows Sandbox reads, and in a container's environment. The
343	343		first thing the runner in the sandbox does is fetch the job with it,
344	344		and g1t answers with a new token that only the sandbox has; the first
345		−	stops working at once, for that and for everything else. A copy left
	345	+	stops working at once for everything that can be asked through g1t's
	346	+	API (inside g1t, only the lookup of [brokered
	347	+	secrets](/guides/secrets-and-variables/#brokered-secrets) still takes it, and that is not reachable
	348	+	from your computer or a job). A copy left
346	349		behind can't fetch the job, its secrets or a `G1T_TOKEN` again. A job's
347	350		folder is deleted when the job ends, and a runner that was stopped
348	351		mid-job (a crash, or `taskkill`) deletes what it left when it next

1	+	-- A self-hosted runner's harness spends the job's token on fetching the
2	+	-- job, and gets a new one in its place (plan.rs `job_spec` with `rotate`).
3	+	-- The token the job started with stops working for everything but
4	+	-- `brokered_secrets`, which the runner Worker asks with the token it was
5	+	-- given at the start and which no API route passes on: its hash is kept
6	+	-- here from the swap until the job ends or is claimed again.
7	+	ALTER TABLE jobs ADD COLUMN first_token_hash TEXT;

186	186		/// `brokered_secrets`: a running job's stand-ins and what they stand
187	187		/// for, for the runner Worker to fill in on the job's requests. Asked
188	188		/// with the job's own token, which only the runner and the job hold;
189		−	/// the API never passes this method on, so the job cannot ask it.
	189	+	/// the API never passes this method on, so the job cannot ask it. The
	190	+	/// Worker holds the token the job started with, so that one still
	191	+	/// answers here after the job's harness swapped it for another
	192	+	/// (`job_spec` with `rotate`), and nowhere else.
190	193		pub async fn brokered_secrets(&self, a: JobCallArgs) -> Result<Outcome<Vec<BrokeredSecret>>> {
191		−	let job = check!(self.job_for_token(&a).await?);
	194	+	let job = check!(self.job_for_started_token(&a).await?);
192	195		Ok(Outcome::Ok(self.stand_ins_of(&job.id).await?))
193	196		}
194	197		}

1212	1212		let claimed = self
1213	1213		.db
1214	1214		.prepare(
1215		−	"UPDATE jobs SET status = 'in_progress', token_hash = ?, runner_id = ?, runner_name = ?, reason = NULL, started_at = ?, seen_at = ?
	1215	+	"UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, runner_id = ?, runner_name = ?, reason = NULL, started_at = ?, seen_at = ?
1216	1216		WHERE id = ? AND status = 'queued' RETURNING id",
1217	1217		)
1218	1218		.bind(&[

flagon-io/g1t

Comments