Skip to content
Sign up

flagon-io/g1t

public
0
380 linesCodeBlameRaw
1//! Telling g1t how a job is going: its steps, its log in batches, its
2//! annotations, and how it ended. Every report carries the job's token.
3
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::time::{Duration, Instant};
6
7use anyhow::Result;
8use g1t_actions::mask;
9use serde_json::{Value, json};
10
11/// How long log lines wait before they are sent.
12const FLUSH_EVERY: Duration = Duration::from_millis(1500);
13/// How much log is sent at once.
14const FLUSH_BYTES: usize = 64 * 1024;
15/// The most one step may add to the job's summary, as on GitHub.
16pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
17/// How long the runner goes without a report before it asks whether the
18/// job was cancelled.
19const PING_EVERY: Duration = Duration::from_secs(10);
20
21/// Whether the job was cancelled, as g1t's answers to its reports say.
22struct Cancel {
23 /// g1t said so.
24 said: AtomicBool,
25 /// The job has taken it in: the step it was on was stopped, and the
26 /// cleanup steps that follow are not stopped for it again.
27 taken: AtomicBool,
28}
29
30impl Cancel {
31 const fn new() -> Cancel {
32 Cancel { said: AtomicBool::new(false), taken: AtomicBool::new(false) }
33 }
34
35 /// Reads an answer to a report: `cancelled` when the run was cancelled.
36 fn hear(&self, answer: &Value) {
37 if answer["cancelled"].as_bool() == Some(true) {
38 self.said.store(true, Ordering::Relaxed);
39 }
40 }
41
42 fn cancelled(&self) -> bool {
43 self.said.load(Ordering::Relaxed)
44 }
45
46 fn interrupt(&self) -> bool {
47 self.cancelled() && !self.taken.load(Ordering::Relaxed)
48 }
49
50 fn take(&self) {
51 self.taken.store(true, Ordering::Relaxed);
52 }
53}
54
55/// This job's (one runs per process).
56static CANCEL: Cancel = Cancel::new();
57
58/// Whether g1t said the job was cancelled.
59pub(crate) fn cancelled() -> bool {
60 CANCEL.cancelled()
61}
62
63/// Whether a running step should be stopped: cancelled, and not yet taken in.
64pub(crate) fn interrupt() -> bool {
65 CANCEL.interrupt()
66}
67
68/// The job has seen the cancellation: what runs from here is its cleanup.
69pub(crate) fn take_cancel() {
70 CANCEL.take();
71}
72
73pub(crate) struct Api {
74 pub(crate) base: String,
75 pub(crate) job: String,
76 pub(crate) token: String,
77}
78
79impl Api {
80 pub(crate) fn spec(&self) -> Result<Value> {
81 let response = ureq::post(&format!("{}/actions/jobs/{}/spec", self.base, self.job))
82 .timeout(Duration::from_secs(60))
83 .send_json(json!({ "token": self.token }))?;
84 Ok(response.into_json()?)
85 }
86
87 /// Where to fetch another repository's action from: `{"source": "g1t",
88 /// "url", "ref", "token"}`, or `{"source": "github", "sha", "archive",
89 /// "note"}`: the commit GitHub's ref names, and whether g1t keeps its
90 /// archive (`action_archive`). Err holds why g1t
91 /// refused (`true`: the repository is private and may not be used
92 /// here), or that it could not be asked (`false`).
93 pub(crate) fn action(&self, repository: &str, git_ref: &str) -> std::result::Result<Value, (bool, String)> {
94 // The first job to use an action at a commit waits while g1t
95 // fetches and keeps its archive.
96 let sent = ureq::post(&format!("{}/actions/jobs/{}/action", self.base, self.job))
97 .timeout(Duration::from_secs(120))
98 .send_json(json!({ "token": self.token, "report": { "repository": repository, "ref": git_ref } }));
99 match sent {
100 Ok(response) => response.into_json().map_err(|error| (false, error.to_string())),
101 Err(ureq::Error::Status(code, response)) => {
102 let body: Value = response.into_json().unwrap_or(Value::Null);
103 let message = body["error"]["message"].as_str().unwrap_or("g1t did not answer.").to_owned();
104 Err((code == 403, message))
105 }
106 Err(error) => Err((false, error.to_string())),
107 }
108 }
109
110 /// Downloads the archive g1t keeps of `repository` at `sha` to `to`.
111 pub(crate) fn action_archive(&self, repository: &str, sha: &str, to: &std::path::Path) -> std::result::Result<u64, String> {
112 let response = ureq::post(&format!("{}/actions/jobs/{}/archive", self.base, self.job))
113 .timeout(Duration::from_secs(300))
114 .send_json(json!({ "token": self.token, "report": { "repository": repository, "sha": sha } }))
115 .map_err(|error| match error {
116 ureq::Error::Status(code, response) => {
117 let body: Value = response.into_json().unwrap_or(Value::Null);
118 body["error"]["message"].as_str().map_or_else(|| format!("g1t answered {code}"), str::to_owned)
119 }
120 other => other.to_string(),
121 })?;
122 let mut file = std::fs::File::create(to).map_err(|error| error.to_string())?;
123 std::io::copy(&mut response.into_reader(), &mut file).map_err(|error| error.to_string())
124 }
125
126 /// An API with no address, which sends nothing: what the tests use, so
127 /// a step's run is not held up by reports that go nowhere (each one
128 /// tried three times, with waits between, over a second and a half).
129 #[cfg(test)]
130 pub(crate) fn offline() -> Api {
131 Api { base: String::new(), job: "job_1".into(), token: "t".into() }
132 }
133
134 pub(crate) fn report(&self, report: Value) {
135 // No address: nowhere to send it (Api::offline).
136 if self.base.is_empty() {
137 return;
138 }
139 // A report that cannot be sent is tried a few times, then dropped:
140 // the job goes on, and g1t notices a silent job by itself.
141 for attempt in 0..3 {
142 let sent = ureq::post(&format!("{}/actions/jobs/{}", self.base, self.job))
143 .timeout(Duration::from_secs(30))
144 .send_json(json!({ "token": self.token, "report": report }));
145 match sent {
146 Ok(response) => {
147 if let Ok(answer) = response.into_json::<Value>() {
148 CANCEL.hear(&answer);
149 }
150 return;
151 }
152 // Refused: the job was cancelled or finished; nothing to retry.
153 Err(ureq::Error::Status(code, _)) if (400..500).contains(&code) => return,
154 Err(_) => std::thread::sleep(Duration::from_millis(500 * (attempt + 1))),
155 }
156 }
157 }
158}
159
160/// The job's log, masked, sent in batches. `masks` holds every form of
161/// every secret (`g1t_actions::mask`), longest first.
162pub(crate) struct Log {
163 pub(crate) api: Api,
164 pub(crate) masks: Vec<String>,
165 step: u32,
166 buffer: String,
167 last: Instant,
168 /// When anything was last sent, for the ping.
169 sent: Instant,
170 /// Everything sent so far, so a test reads the whole log however its
171 /// batches fell (`buffered`).
172 #[cfg(test)]
173 flushed: String,
174}
175
176impl Log {
177 pub(crate) fn new(api: Api, mut masks: Vec<String>) -> Log {
178 masks.retain(|mask| !mask.is_empty());
179 masks.sort_by_key(|mask| std::cmp::Reverse(mask.len()));
180 Log {
181 api,
182 masks,
183 step: 0,
184 buffer: String::new(),
185 last: Instant::now(),
186 sent: Instant::now(),
187 #[cfg(test)]
188 flushed: String::new(),
189 }
190 }
191
192 /// The log so far: what was sent and what waits to be.
193 #[cfg(all(test, unix))]
194 pub(crate) fn buffered(&self) -> String {
195 format!("{}{}", self.flushed, self.buffer)
196 }
197
198 /// Starts writing to step `number` (0 for the job's setup).
199 pub(crate) fn step(&mut self, number: u32) {
200 self.flush();
201 self.step = number;
202 }
203
204 pub(crate) fn mask(&self, text: &str) -> String {
205 mask::apply(text, &self.masks)
206 }
207
208 /// `::add-mask::`: masks `value` from here on, in every form it can
209 /// take (each line, base64, JSON-escaped), as a secret is.
210 pub(crate) fn add_mask(&mut self, value: &str) {
211 let mut added = false;
212 for variant in mask::variants(value) {
213 if !self.masks.contains(&variant) {
214 self.masks.push(variant);
215 added = true;
216 }
217 }
218 if added {
219 self.masks.sort_by_key(|mask| std::cmp::Reverse(mask.len()));
220 }
221 }
222
223 pub(crate) fn line(&mut self, text: &str) {
224 let masked = self.mask(text);
225 self.buffer.push_str(&masked);
226 self.buffer.push('\n');
227 if self.buffer.len() >= FLUSH_BYTES || self.last.elapsed() >= FLUSH_EVERY {
228 self.flush();
229 }
230 }
231
232 /// Sends what is waiting if it has waited long enough, and asks after
233 /// the job when nothing has been sent for a while, so a cancellation
234 /// reaches a step that prints nothing.
235 pub(crate) fn tick(&mut self) {
236 if !self.buffer.is_empty() && self.last.elapsed() >= FLUSH_EVERY {
237 self.flush();
238 } else if self.sent.elapsed() >= PING_EVERY {
239 self.sent = Instant::now();
240 self.api.report(json!({ "kind": "ping" }));
241 }
242 }
243
244 pub(crate) fn flush(&mut self) {
245 self.last = Instant::now();
246 if self.buffer.is_empty() {
247 return;
248 }
249 self.sent = Instant::now();
250 let text = std::mem::take(&mut self.buffer);
251 #[cfg(test)]
252 self.flushed.push_str(&text);
253 self.api.report(json!({ "kind": "log", "step": self.step, "text": text }));
254 }
255
256 pub(crate) fn steps(&self, names: &[String]) {
257 self.api.report(json!({ "kind": "steps", "steps": names }));
258 }
259
260 pub(crate) fn step_state(&mut self, number: u32, name: &str, status: &str, conclusion: Option<&str>) {
261 self.flush();
262 let name = self.mask(name);
263 self.api.report(json!({ "kind": "step", "number": number, "name": name, "status": status, "conclusion": conclusion }));
264 }
265
266 /// What a step wrote to `$GITHUB_STEP_SUMMARY`, masked, for the run's
267 /// page. More than 1 MiB is refused, as on GitHub, with an error in the
268 /// log.
269 pub(crate) fn summary(&mut self, markdown: &str) {
270 if markdown.trim().is_empty() {
271 return;
272 }
273 if markdown.len() > MAX_SUMMARY_BYTES {
274 self.line(&format!(
275 "##[error]$GITHUB_STEP_SUMMARY upload aborted: a step's summary may be up to 1024k, and this one is {}k.",
276 markdown.len().div_ceil(1024)
277 ));
278 return;
279 }
280 let markdown = self.mask(markdown);
281 self.flush();
282 self.api.report(json!({ "kind": "summary", "step": self.step, "markdown": markdown }));
283 }
284
285 pub(crate) fn annotation(&mut self, level: &str, message: &str, properties: &serde_json::Map<String, Value>) {
286 let message = self.mask(message);
287 let title = properties.get("title").and_then(Value::as_str).map(|title| self.mask(title));
288 self.api.report(json!({
289 "kind": "annotation",
290 "level": level,
291 "message": message,
292 "title": title,
293 "file": properties.get("file"),
294 "line": properties.get("line").and_then(|l| l.as_str()).and_then(|l| l.parse::<u32>().ok()),
295 }));
296 }
297
298 pub(crate) fn done(&mut self, conclusion: &str, outputs: &serde_json::Map<String, Value>, reason: Option<&str>) {
299 let outputs = self.withhold_secrets(outputs);
300 self.flush();
301 self.api.report(json!({ "kind": "done", "conclusion": conclusion, "outputs": outputs, "reason": reason }));
302 }
303
304 /// The job's outputs without any that hold a secret, as on GitHub: an
305 /// output goes to other jobs and to the run's page, where no mask
306 /// reaches. Each one left out is warned of in the log.
307 pub(crate) fn withhold_secrets(&mut self, outputs: &serde_json::Map<String, Value>) -> serde_json::Map<String, Value> {
308 let mut kept = serde_json::Map::new();
309 for (name, value) in outputs {
310 let text = match value {
311 Value::String(text) => text.clone(),
312 other => other.to_string(),
313 };
314 if mask::reveals(&text, &self.masks) {
315 self.line(&format!("##[warning]The output `{name}` was left out: it holds a secret."));
316 continue;
317 }
318 kept.insert(name.clone(), value.clone());
319 }
320 kept
321 }
322}
323
324#[cfg(test)]
325mod tests {
326 use super::*;
327
328 fn log(secrets: &[&str]) -> Log {
329 Log::new(Api::offline(), mask::all_variants(secrets.iter().copied()))
330 }
331
332 #[test]
333 fn outputs_holding_a_secret_are_left_out() {
334 let mut log = log(&["s3cr3t-token"]);
335 let mut outputs = serde_json::Map::new();
336 outputs.insert("version".into(), json!("1.2.0"));
337 outputs.insert("leak".into(), json!("token=s3cr3t-token"));
338 outputs.insert("encoded".into(), json!("czNjcjN0LXRva2Vu"));
339 let kept = log.withhold_secrets(&outputs);
340 assert_eq!(kept.keys().collect::<Vec<_>>(), ["version"]);
341 assert!(log.buffer.contains("The output `leak` was left out"));
342 assert!(log.buffer.contains("The output `encoded` was left out"));
343 }
344
345 #[test]
346 fn a_cancelled_answer_is_heard_once_and_taken_in() {
347 let cancel = Cancel::new();
348 cancel.hear(&json!({ "ok": true }));
349 assert!(!cancel.cancelled());
350 cancel.hear(&json!({ "ok": true, "cancelled": true }));
351 assert!(cancel.cancelled() && cancel.interrupt());
352 cancel.take();
353 assert!(cancel.cancelled() && !cancel.interrupt(), "cleanup steps are not stopped again");
354 cancel.hear(&json!({ "ok": true, "cancelled": false }));
355 assert!(cancel.cancelled(), "a cancelled job stays cancelled");
356 }
357
358 #[test]
359 fn a_summary_past_its_limit_is_refused_in_the_log() {
360 let mut log = log(&[]);
361 log.summary("
362");
363 assert!(log.buffer.is_empty(), "an empty summary is not sent");
364 log.summary(&"x".repeat(MAX_SUMMARY_BYTES + 1));
365 assert!(log.buffer.contains("$GITHUB_STEP_SUMMARY upload aborted"));
366 assert!(log.buffer.contains("1025k"));
367 }
368
369 #[test]
370 fn added_masks_cover_every_form() {
371 let mut log = log(&[]);
372 log.add_mask("line one\nline two");
373 assert_eq!(log.mask("first: line one"), "first: ***");
374 assert_eq!(log.mask("then line two"), "then ***");
375 // Longest first: the whole value, not its pieces.
376 log.add_mask("abc");
377 log.add_mask("abcdef");
378 assert_eq!(log.mask("abcdef"), "***");
379 }
380}