format_summary
def format_summary(
rep, verb:str='ran'
):The one line that replaces a message per submodule. Quiet submodules are counted, not narrated
Fleet
Run one command across every submodule of a superproject
Discovery and reporting
Every verb walks the same list and reports the same way.
Exported source
import json
import re
import shutil
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path
from fastcore.script import *
# The rendering layer lives in `automate`, the git/discovery layer in `github`. This
# module is the only one that knows about MANY repos; those two stay single-repo.
from nbdevAuto.automate import StepFailed, _push_summary, _run, _step, console, escape
from nbdevAuto.github import find_root, git, local_state, parse_gitmodulesExported source
class Skip(Exception):
"Raised by a walked callable that has decided this submodule has nothing to do"
@dataclass
class Outcome:
"What happened in one submodule"
path: str
state: str # ran | skipped | failed
reason: str = "" # why it was skipped, or how it failed
detail: list = field(default_factory=list) # short lines to show under the path
@dataclass
class Report:
"Everything a `walk` did, and the exit code it implies"
outcomes: list = field(default_factory=list)
@property
def ran(self): return [o for o in self.outcomes if o.state == "ran"]
@property
def failed(self): return [o for o in self.outcomes if o.state == "failed"]
@property
def code(self): return 1 if self.failed else 0
def skip_counts(self):
"Skip reason -> count, in the order the reasons were first seen"
counts = {}
for o in self.outcomes:
if o.state == "skipped": counts[o.reason] = counts.get(o.reason, 0) + 1
return counts
def format_summary(rep, verb="ran"):
"The one line that replaces a message per submodule. Quiet submodules are counted, not narrated"
n = len(rep.outcomes)
parts = [f"{len(rep.ran)} {verb}"]
parts += [f"{c} {reason}" for reason, c in rep.skip_counts().items()]
if rep.failed: parts.append(f"{len(rep.failed)} FAILED")
return f"{n} submodule{'' if n == 1 else 's'}: " + ", ".join(parts)Report
def Report(
outcomes:list=<factory>
)->None:Everything a walk did, and the exit code it implies
Outcome
def Outcome(
path:str, state:str, reason:str='', detail:list=<factory>
)->None:What happened in one submodule
Skip
def Skip(
*args, **kwargs
):Raised by a walked callable that has decided this submodule has nothing to do
finish
def finish(
rep, verb
):Print the counted summary, name any failures, and return the exit code
Exported source
def _classify(base, sub, need_nbs, only_dirty):
"The reason to skip this submodule before running anything, or None to run it"
p = base/sub.path
# An uninitialised submodule is an empty directory: present, but with no .git
if not (p/".git").exists(): return "not initialised"
if need_nbs and not (p/"nbs").is_dir(): return "without nbs"
if only_dirty and not git(p, "status", "--porcelain"): return "clean"
return None
def _fail_parts(e):
"""(headline, remaining lines) for a callable that did not finish.
A `StepFailed` carries the command's own output, and its first line is the message
worth putting on the summary line - "uncommitted changes" says far more than
"update exited 1". The rest is kept for the block renderer."""
if isinstance(e, StepFailed):
out = (e.output or "").strip().splitlines()
if out: return out[0], out[1:]
return f"{e.cmd} exited {e.code}", []
return f"{type(e).__name__}: {e}", []
def _render(o, style="block", width=0, header=True):
"One submodule's result. A skip prints nothing at all; it is counted instead"
if o.state == "skipped": return
if style == "line":
pad = f"{o.path:<{width}}"
if o.state == "failed":
console.print(f"[bold red]{escape(pad)}[/bold red] [red]{escape(o.reason)}[/red]")
return
console.print(f"[cyan]{escape(pad)}[/cyan] [dim]{escape(o.detail[0] if o.detail else '')}[/dim]")
for line in o.detail[1:]: console.print(f"{' ' * width} [dim]{escape(line)}[/dim]")
return
if header: console.print(f"\n[bold cyan]==> {escape(o.path)}[/bold cyan]")
if o.state == "failed":
console.print(f" [bold red]FAIL[/bold red] [red]{escape(o.reason)}[/red]")
for line in o.detail: console.print(f" [dim]{escape(line)}[/dim]")
def walk(fn, *, root=None, only_dirty=False, need_nbs=False, workers=1, style="block"):
"""Run `fn(path, sub)` in each submodule, returning a `Report`.
`fn` returns a list of short strings to show under the submodule's name, or raises
`Skip` once it knows there is nothing to do. Only submodules that do something print;
the rest are counted by `format_summary`.
`style="block"` gives each submodule a `==> path` heading with its own output beneath,
which is what a long command like `upload` needs. `style="line"` gives one padded line
per submodule, which suits verbs that report a single fact.
Sequential block runs print the heading BEFORE calling `fn`, so its output streams
underneath. Everything else prints after, in submodule order, so parallel work cannot
interleave and two runs of the same fleet read the same way."""
base = Path(root) if root else find_root()
rep, todo = Report(), []
for s in parse_gitmodules(base):
why = _classify(base, s, need_nbs, only_dirty)
if why: rep.outcomes.append(Outcome(s.path, "skipped", why))
else: todo.append(s)
width = max((len(s.path) for s in todo), default=0) + 2
def _one(s):
try: return Outcome(s.path, "ran", detail=list(fn(base/s.path, s) or []))
except Skip as e: return Outcome(s.path, "skipped", str(e))
except BaseException as e:
head, rest = _fail_parts(e)
return Outcome(s.path, "failed", head, rest)
if workers > 1 and len(todo) > 1:
with ThreadPoolExecutor(max_workers=workers) as ex: done = list(ex.map(_one, todo))
for o in done: _render(o, style, width)
else:
done = []
live = style == "block"
for s in todo:
if live: console.print(f"\n[bold cyan]==> {escape(s.path)}[/bold cyan]")
o = _one(s)
done.append(o)
_render(o, style, width, header=not live)
rep.outcomes += done
rep.outcomes.sort(key=lambda o: o.path)
return rep
def finish(rep, verb):
"Print the counted summary, name any failures, and return the exit code"
console.print(f"\n[bold]{escape(format_summary(rep, verb))}[/bold]")
for o in rep.failed:
console.print(f"[bold red]FAILED[/bold red] {escape(o.path)}: [dim]{escape(o.reason)}[/dim]")
return rep.codewalk
def walk(
fn, *, root:NoneType=None, only_dirty:bool=False, need_nbs:bool=False, workers:int=1, style:str='block'
):Run fn(path, sub) in each submodule, returning a Report.
fn returns a list of short strings to show under the submodule’s name, or raises Skip once it knows there is nothing to do. Only submodules that do something print; the rest are counted by format_summary.
style="block" gives each submodule a ==> path heading with its own output beneath, which is what a long command like upload needs. style="line" gives one padded line per submodule, which suits verbs that report a single fact.
Sequential block runs print the heading BEFORE calling fn, so its output streams underneath. Everything else prints after, in submodule order, so parallel work cannot interleave and two runs of the same fleet read the same way.
Running a console script per submodule
upload_one
def upload_one(
path, sub:NoneType=None, args:tuple=()
):Run the single-repo upload in path, letting its own rich output through.
A subprocess rather than an in-process call: prep reads nbdev’s config from the cwd and caches it, and nbdev_test spawns its own process pool, so nineteen repos in one interpreter is a trap. The subprocess keeps this tty, so colour survives.
Exported source
def script_path(name):
"""The console script `name` from the environment this is running in.
Taking it from `sys.executable`'s directory is what makes this work under a
non-interactive shell that never put the venv on PATH."""
cand = Path(sys.executable).parent/name
if cand.is_file(): return str(cand)
found = shutil.which(name)
if found: return found
raise RuntimeError(f"Cannot find the '{name}' console script. Run `uv sync` first")
def upload_one(path, sub=None, args=()):
"""Run the single-repo `upload` in `path`, letting its own rich output through.
A subprocess rather than an in-process call: `prep` reads nbdev's config from the cwd
and caches it, and `nbdev_test` spawns its own process pool, so nineteen repos in one
interpreter is a trap. The subprocess keeps this tty, so colour survives."""
r = subprocess.run([script_path("upload"), *args], cwd=str(path), check=False)
if r.returncode: raise StepFailed("upload", r.returncode)
return []script_path
def script_path(
name
):The console script name from the environment this is running in.
Taking it from sys.executable’s directory is what makes this work under a non-interactive shell that never put the venv on PATH.
Verbs
fleet_upload
def fleet_upload(
all:bool=False, # Upload every submodule with an nbs/ folder, not just the changed ones
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Run upload in each submodule that has local changes
Exported source
def fleet_upload(
all:bool = False, # Upload every submodule with an nbs/ folder, not just the changed ones
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Run `upload` in each submodule that has local changes"
# Sequential on purpose: `upload` runs nbdev_test with 8 workers, and knowledge-lab
# has 4 vCPU. Parallel uploads would oversubscribe the box and interleave the output.
rep = walk(upload_one, root=root, need_nbs=True, only_dirty=not all, workers=1)
sys.exit(finish(rep, "uploaded"))push and copy. Both report one line per submodule, and both take --dry-run.
copy_one
def copy_one(
path, sub:NoneType=None, files:tuple=(), dest:str='nbs', dry_run:bool=False
):Copy files into path/dest.
The file list belongs to the caller, so this stays a general-purpose copy rather than learning about any particular superproject’s shared chrome.
Exported source
def ahead_of_remote(path):
"(commits origin has not seen, branch name). Branch is None when HEAD is detached"
br = git(path, "rev-parse", "--abbrev-ref", "HEAD")
if not br or br == "HEAD": return 0, None
n = git(path, "rev-list", "--count", f"origin/{br}..HEAD")
return (int(n) if n.isdigit() else 0), br
def push_one(path, sub=None, dry_run=False):
"Push `path` when it has local commits origin has not seen"
n, br = ahead_of_remote(path)
if br is None: raise Skip("detached HEAD")
if not git(path, "remote"): raise Skip("no remote")
if not n: raise Skip("nothing to push")
if dry_run: return [f"would push {n} commit{'' if n == 1 else 's'} on {br}"]
return _push_summary(_run("git", "-C", str(path), "push"))
def copy_one(path, sub=None, files=(), dest="nbs", dry_run=False):
"""Copy `files` into `path`/`dest`.
The file list belongs to the caller, so this stays a general-purpose copy rather than
learning about any particular superproject's shared chrome."""
out = Path(path)/dest
if not out.is_dir(): raise Skip(f"no {dest}/")
names = []
for f in files:
src = Path(f)
if not src.is_file(): raise StepFailed(f"copy {src.name}", 1, f"no such file: {src}")
if not dry_run: shutil.copy2(src, out/src.name)
names.append(src.name)
return [("would copy " if dry_run else "copied ") + ", ".join(names) + f" -> {dest}/"]push_one
def push_one(
path, sub:NoneType=None, dry_run:bool=False
):Push path when it has local commits origin has not seen
ahead_of_remote
def ahead_of_remote(
path
):(commits origin has not seen, branch name). Branch is None when HEAD is detached
fleet_copy
def fleet_copy(
files:str, # Files to copy into each submodule, comma or space separated
dest:str='nbs', # Directory inside each submodule to copy into
dry_run:bool=False, # Report what would be copied without copying it
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Copy shared files into every submodule that has the destination directory
Exported source
def fleet_push(
dry_run:bool = False, # Report what would be pushed without pushing it
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Push every submodule that has local commits not yet on its remote"
# Network-bound and independent per remote, so this one does parallelise.
rep = walk(lambda p, s: push_one(p, s, dry_run=dry_run),
root=root, workers=8, style="line")
sys.exit(finish(rep, "would push" if dry_run else "pushed"))
def fleet_copy(
files:str, # Files to copy into each submodule, comma or space separated
dest:str = "nbs", # Directory inside each submodule to copy into
dry_run:bool = False, # Report what would be copied without copying it
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Copy shared files into every submodule that has the destination directory"
fs = [f for f in re.split(r"[,\s]+", files) if f]
if not fs: sys.exit("fleet copy: no files given")
rep = walk(lambda p, s: copy_one(p, s, files=fs, dest=dest, dry_run=dry_run),
root=root, workers=8, style="line")
sys.exit(finish(rep, "would update" if dry_run else "updated"))fleet_push
def fleet_push(
dry_run:bool=False, # Report what would be pushed without pushing it
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Push every submodule that has local commits not yet on its remote
status replaces both the git submodule status loop and the gss recipe.
fleet_status
def fleet_status(
short:bool=False, # Only the submodules with changes, listing the changed paths
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Branch, dirty flag and HEAD for every submodule
Exported source
def _porcelain(path):
"""`git status -s` for `path`, NOT stripped.
`github.git` strips its output, and `git status -s` pads the code to two columns, so
stripping eats the leading space of the FIRST line and shifts that one path by a
character. `automate.status` documents the same trap."""
return subprocess.run(["git", "-C", str(path), "status", "-s"],
capture_output=True, text=True, check=False).stdout
def status_one(path, sub=None, short=False):
"""Branch, dirty flag and short HEAD for `path`.
With `short`, clean submodules are skipped and the dirty ones list their changed
paths - the fleet-wide "what have I got outstanding" view."""
st = local_state(path)
br = st["branch"] or "?"
if br == "HEAD": br = "DETACHED"
line = f"{br:<10} {'dirty' if st['dirty'] else 'clean':<6} {st['head'] or '-'}"
if not short: return [line]
if not st["dirty"]: raise Skip("clean")
return [line] + [f" {l}" for l in _porcelain(path).splitlines()]
def fleet_status(
short:bool = False, # Only the submodules with changes, listing the changed paths
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Branch, dirty flag and HEAD for every submodule"
rep = walk(lambda p, s: status_one(p, s, short=short), root=root, workers=8, style="line")
sys.exit(finish(rep, "changed" if short else "reported"))status_one
def status_one(
path, sub:NoneType=None, short:bool=False
):Branch, dirty flag and short HEAD for path.
With short, clean submodules are skipped and the dirty ones list their changed paths - the fleet-wide “what have I got outstanding” view.
update and sync, the destructive pair. Fetching parallelises; checkout and merge do not.
pull_root
def pull_root(
base
):Fast-forward the superproject itself, with the advice its usual failure needs
Exported source
def _git_ok(path, *args):
"True when git succeeded in `path`. `git()` hides the exit code, and here it matters"
return subprocess.run(["git", "-C", str(path), *args],
capture_output=True, check=False).returncode == 0
def _count(path, rng):
"Commit count for an `a..b` range, 0 when the range does not resolve"
n = git(path, "rev-list", "--count", rng)
return int(n) if n.isdigit() else 0
def fetch_all(root=None, workers=8):
"Fetch every initialised submodule concurrently. This is the slow half of an update"
base = Path(root) if root else find_root()
subs = [s for s in parse_gitmodules(base) if (base/s.path/".git").exists()]
with ThreadPoolExecutor(max_workers=workers) as ex:
list(ex.map(lambda s: git(base/s.path, "fetch", "-q", "origin"), subs))
return len(subs)
def update_one(path, sub=None, dry_run=False, branch="main"):
"""Reattach `path` to `branch` and fast-forward it to origin/<branch>.
Assumes `fetch_all` already ran: fetching is the half that parallelises, and this is
the half that must not, because a checkout rewrites the worktree."""
if git(path, "status", "--porcelain"):
raise StepFailed("update", 1, "uncommitted changes; commit or stash first")
if not git(path, "remote"): raise Skip("no remote")
target = f"origin/{branch}"
if not git(path, "rev-parse", "--verify", "--quiet", target): raise Skip(f"no {target}")
br = git(path, "rev-parse", "--abbrev-ref", "HEAD")
move = br != branch
# Compare the branch we are ABOUT to be on, not the one we happen to be on now
ref = branch if _git_ok(path, "rev-parse", "--verify", "--quiet", branch) else "HEAD"
behind = _count(path, f"{ref}..{target}")
ahead = _count(path, f"{target}..{ref}")
# Diverged needs a human. The bash version echoed a line and moved on, which is easy
# to scroll past; this fails the run so the exit code carries it.
if ahead and behind:
raise StepFailed("merge --ff-only", 1,
f"diverged from {target}: {ahead} ahead, {behind} behind - resolve by hand")
was = "DETACHED" if br == "HEAD" else br
if not move and not behind: raise Skip("up to date")
if dry_run:
bits = ([f"checkout {branch} (from {was})"] if move else []) + \
([f"fast-forward {behind}"] if behind else [])
return ["would " + ", ".join(bits)]
done = []
if move:
if not _git_ok(path, "checkout", "-q", branch):
_run("git", "-C", str(path), "checkout", "-q", "-b", branch, target)
done.append(f"checked out {branch} (was {was})")
behind = _count(path, f"HEAD..{target}") # the checkout may already have moved us
if behind:
_run("git", "-C", str(path), "merge", "--ff-only", target)
done.append(f"fast-forwarded {behind} commit{'' if behind == 1 else 's'}")
return done
def dirty_submodules(root=None):
"Submodule paths holding uncommitted changes"
base = Path(root) if root else find_root()
return [s.path for s in parse_gitmodules(base)
if (base/s.path/".git").exists() and git(base/s.path, "status", "--porcelain")]
def init_new(root=None, dry_run=False):
"""Initialise only the submodules git reports as uninitialised.
A blanket `submodule update --init` would reset every initialised submodule back to
its recorded gitlink and detach its HEAD, which is the one thing this setup avoids."""
base = Path(root) if root else find_root()
new = [l.split()[1] for l in git(base, "submodule", "status").splitlines() if l.startswith("-")]
if new and not dry_run:
_run("git", "-C", str(base), "submodule", "update", "--init", "--", *new)
return new
def pull_root(base):
"Fast-forward the superproject itself, with the advice its usual failure needs"
try: return _run("git", "-C", str(base), "pull", "--ff-only").strip().splitlines()[-1:]
except StepFailed as e:
raise StepFailed(e.cmd, e.code, f"{e.output}\n"
"If it complains about submodule paths, the recorded gitlinks drifted.\n"
"Discard them (the update step re-advances them anyway):\n"
" git checkout -- $(git diff --name-only)") from Noneinit_new
def init_new(
root:NoneType=None, dry_run:bool=False
):Initialise only the submodules git reports as uninitialised.
A blanket submodule update --init would reset every initialised submodule back to its recorded gitlink and detach its HEAD, which is the one thing this setup avoids.
dirty_submodules
def dirty_submodules(
root:NoneType=None
):Submodule paths holding uncommitted changes
update_one
def update_one(
path, sub:NoneType=None, dry_run:bool=False, branch:str='main'
):Reattach path to branch and fast-forward it to origin/
Assumes fetch_all already ran: fetching is the half that parallelises, and this is the half that must not, because a checkout rewrites the worktree.
fetch_all
def fetch_all(
root:NoneType=None, workers:int=8
):Fetch every initialised submodule concurrently. This is the slow half of an update
fleet_sync
def fleet_sync(
dry_run:bool=False, # Report what would move without moving it
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Bring this checkout level with every remote: pull, init new submodules, fast-forward each
Exported source
def fleet_update(
dry_run:bool = False, # Report what would move without moving it
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Fetch every submodule, then fast-forward each one's main to origin/main"
base = Path(root) if root else find_root()
with _step(1, 2, "fetch every submodule") as d:
d += [f"{fetch_all(base)} fetched"]
console.print("[dim]2/2[/dim] [bold]fast-forward each main[/bold]")
rep = walk(lambda p, s: update_one(p, s, dry_run=dry_run), root=base, workers=1, style="line")
sys.exit(finish(rep, "would update" if dry_run else "updated"))
def fleet_sync(
dry_run:bool = False, # Report what would move without moving it
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Bring this checkout level with every remote: pull, init new submodules, fast-forward each"
base = Path(root) if root else find_root()
# Pre-flight FIRST: a dirty submodule turns the checkout/merge below into a conflicted
# tree, so nothing at all should happen until this passes.
dirty = dirty_submodules(base)
if dirty:
console.print("[bold red]Uncommitted changes in:[/bold red]")
for d in dirty: console.print(f" {escape(d)}")
console.print("Commit or stash them first ([bold]just upload[/bold]), then re-run.")
sys.exit(1)
with _step(1, 3, "pull the superproject") as d:
d += ["skipped (dry run)"] if dry_run else pull_root(base)
with _step(2, 3, "initialise new submodules") as d:
new = init_new(base, dry_run=dry_run)
d += [f"{'would init ' if dry_run else 'initialised '}{', '.join(new)}"] if new else ["none"]
with _step(3, 3, "fetch every submodule") as d:
d += [f"{fetch_all(base)} fetched"]
rep = walk(lambda p, s: update_one(p, s, dry_run=dry_run), root=base, workers=1, style="line")
sys.exit(finish(rep, "would update" if dry_run else "updated"))fleet_update
def fleet_update(
dry_run:bool=False, # Report what would move without moving it
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Fetch every submodule, then fast-forward each one’s main to origin/main
Scanning for personal data
scan is the guardrail for issue #18: personal data reaching a published site. It is the one verb that reads file contents rather than asking git a question, so it is deliberately narrow.
Two things it knows that a plain grep does not. A notebook’s stored outputs are published too, so it scans outputs as well as source - a cleaned-up code cell whose stale output still holds the old figures is exactly the leak that prompted the issue. And it separates published from merely tracked, because the project.render exclude list in _quarto.yml drops p_* notebooks from the site but not from the repo.
scan_file
def scan_file(
path
):Pattern names found in path, or [] for nothing. Non-text files are never scanned
Exported source
# A finding's severity is which of these two lists a file falls in. `_quarto.yml`
# renders every notebook EXCEPT `p_*`, so "published" is a filename test, not a
# guess: a hit in a published file is live on the web, a hit in a `p_` one is
# only in git.
RAW_DOC_SUFFIXES = {".pdf", ".csv", ".ods", ".xlsx", ".xls", ".db", ".sqlite3"}
# Vendor prefixes that are a secret by construction, whatever they are assigned to.
KNOWN_TOKEN = re.compile(
r"\b(?:AKIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9]{20,}|glpat-[A-Za-z0-9_-]{20,}"
r"|sk-[A-Za-z0-9]{20,}|xox[baprs]-[A-Za-z0-9-]{10,})\b")
SECRET_PATTERNS = [
# The header alone is not a key. Every VPN and SSH page in the fleet shows the
# armour wrapped around a "(Insert Client Key)" placeholder, so real base64
# key material has to follow before this counts.
("private key", re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----\s*[\\n\s]*"
r"[A-Za-z0-9+/]{40,}")),
("known token", KNOWN_TOKEN),
("password hash", re.compile(r"\b(?:pbkdf2_sha256|argon2[a-z]*)\$"
r"[^\s\"']*\$[A-Za-z0-9+/]{16,}")),
("tax file no", re.compile(r"(?i)\btax file number\b[^\n]{0,20}?\b\d{8,9}\b")),
("passport no", re.compile(r"(?i)\bpassport\b[^\n]{0,30}?\b[A-Z]{1,2}\d{7}\b")),
("hex secret", re.compile(r"(?i)\b(?:secret|token|api[_ -]?key|client[_ -]?id)\b"
r"\W{0,4}\b([0-9a-f]{32,})\b")),
("credential", re.compile(r"(?i)\b(?:password|passwd|api[_ -]?key|secret|token)\b"
r"\s*[:=]\s*[\"']?([^\s\"',}{]{8,})")),
]
def looks_secret(v):
"""Is this captured value a literal secret, or just code that mentions one?
The credential regex cannot tell `token = 'b4bTbQ...'` from
`password = forms.CharField(...)`, and the second is what every Django,
Stripe and Grafana page in the fleet is FULL of. Ruling those out is the
whole difference between a scan worth running and one everybody ignores.
So the value has to survive four filters: be a plain literal rather than an
expression, not be an identifier or env-var name, not already be masked, and
mix character classes the way a generated secret does and a word does not."""
if len(v) < 16: return False
if not re.fullmatch(r"[A-Za-z0-9_\-+/=.~]+", v): return False # an expression
if re.fullmatch(r"[A-Z0-9_]+", v): return False # ENV_VAR_NAME
if re.fullmatch(r"[A-Za-z_][\w.]*", v) and "." in v:return False # settings.API_KEY
if re.search(r"[*x]{4,}|\.{3,}", v): return False # already masked
classes = sum(bool(re.search(c, v)) for c in (r"[a-z]", r"[A-Z]", r"\d"))
return classes >= 3
# An escape hatch for the deliberately secret-shaped: this module's own test
# vectors, and a doc that has to show what a real token looks like. Put the
# marker on the same line. It is a line comment, so it survives `nbdev-export`.
ALLOW = "scan:allow"
def _line_of(text, pos):
"The whole line containing offset `pos`, so an inline marker can be seen"
return text[text.rfind("\n", 0, pos) + 1 : (text.find("\n", pos) + 1 or None)]
def _hits(text):
"Every secret-pattern name matching `text`, in pattern order, without duplicates"
out = []
for name, rx in SECRET_PATTERNS:
for m in rx.finditer(text):
if ALLOW in _line_of(text, m.start()): continue
# A pattern with a group captures the VALUE, so it can be judged.
# A pattern without one matched something already unambiguous.
if m.groups() and name == "credential" and not looks_secret(m.group(1)): continue
out.append(name); break
return out
def notebook_text(path):
"""Cell sources plus TEXT outputs, with image payloads dropped.
Scanning outputs is the point, not an extra: a code cell rewritten to drop its
figures still carries the OLD numbers in its stored output, and Quarto
publishes that output verbatim. That is the exact leak issue #18 found.
Skipping `image/*` is not just speed. A notebook of plots is mostly base64,
and feeding megabytes of it to these regexes manufactures false positives out
of pure entropy."""
try: nb = json.loads(path.read_text(encoding="utf-8", errors="replace"))
except Exception: return ""
parts = []
for c in nb.get("cells", []):
parts.append("".join(c.get("source", [])))
for o in c.get("outputs", []):
parts.append("".join(o.get("text", [])))
for mime, val in (o.get("data") or {}).items():
if not mime.startswith("image/"):
parts.append("".join(val) if isinstance(val, list) else str(val))
return "\n".join(parts)
# Rendered by `_quarto.yml`, so a hit here is on the public web. Everything else
# tracked is only in git, which is a different and smaller problem.
RENDERED_SUFFIXES = {".ipynb", ".md"}
# Read for content. A binary document cannot be scanned without dragging in a
# parser, so it gets counted instead of judged.
TEXT_SUFFIXES = RENDERED_SUFFIXES | {".py", ".yml", ".yaml", ".txt", ".json", ".csv",
".ini", ".cfg", ".toml", ".sh", ".env"}
EMAIL = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
# RFC 2606 reserves these for documentation, so a page full of them is a tutorial,
# not a contact list. Without this every mail and auth page in the fleet reports.
RESERVED_DOMAINS = re.compile(r"(?i)@(?:[\w.-]+\.)?(?:example\.(?:com|org|net)|"
r"test|localhost|invalid|example)$")
# One address is a contact detail; a column of them is somebody's user table.
BULK_EMAIL_MIN = 5
def is_published(path):
"Does the site render this file? `p_*` is excluded by `project.render` in `_quarto.yml`"
return path.suffix in RENDERED_SUFFIXES and not path.name.startswith("p_")
def scan_file(path):
"Pattern names found in `path`, or [] for nothing. Non-text files are never scanned"
if path.suffix.lower() not in TEXT_SUFFIXES: return []
text = notebook_text(path) if path.suffix == ".ipynb" else \
path.read_text(encoding="utf-8", errors="replace")
names = _hits(text)
addrs = {m.group(0).lower() for m in EMAIL.finditer(text)
if not RESERVED_DOMAINS.search(m.group(0))}
if len(addrs) >= BULK_EMAIL_MIN: names.append(f"{len(addrs)} email addresses")
return namesis_published
def is_published(
path
):Does the site render this file? p_* is excluded by project.render in _quarto.yml
notebook_text
def notebook_text(
path
):Cell sources plus TEXT outputs, with image payloads dropped.
Scanning outputs is the point, not an extra: a code cell rewritten to drop its figures still carries the OLD numbers in its stored output, and Quarto publishes that output verbatim. That is the exact leak issue #18 found.
Skipping image/* is not just speed. A notebook of plots is mostly base64, and feeding megabytes of it to these regexes manufactures false positives out of pure entropy.
looks_secret
def looks_secret(
v
):Is this captured value a literal secret, or just code that mentions one?
The credential regex cannot tell token = 'b4bTbQ...' from password = forms.CharField(...), and the second is what every Django, Stripe and Grafana page in the fleet is FULL of. Ruling those out is the whole difference between a scan worth running and one everybody ignores.
So the value has to survive four filters: be a plain literal rather than an expression, not be an identifier or env-var name, not already be masked, and mix character classes the way a generated secret does and a word does not.
fleet_scan
def fleet_scan(
detail:bool=False, # List every finding, not just the per-submodule counts
strict:bool=False, # Exit 1 on any finding, not only on a published one
root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):Find personal data and credentials in tracked files, published ones first
Exported source
def scan_one(path, sub=None, detail=False, nbs="nbs"):
"""Secret and personal-data findings for one submodule.
Only TRACKED files are scanned. An untracked file is on this disk and nowhere
else, which is exactly where `.gitignore` is supposed to put a payslip, so
flagging it would punish the fix.
Binary documents are counted rather than read. The count is informational and
never fails the scan: a tracked PDF may be a payslip or may be a datasheet,
and this cannot tell without a parser it has no business importing."""
root = path/nbs
if not root.is_dir(): raise Skip("without nbs")
tracked = [l for l in git(path, "ls-files", "-z", nbs).split("\0") if l]
pub, tracked_hits, docs = [], [], 0
for rel in sorted(tracked):
f = path/rel
if not f.is_file(): continue
if f.suffix.lower() in RAW_DOC_SUFFIXES and f.suffix.lower() != ".csv":
docs += 1; continue
names = scan_file(f)
if not names: continue
(pub if is_published(f) else tracked_hits).append(f"{rel}: {', '.join(names)}")
if not (pub or tracked_hits): raise Skip("clean")
line = f"{len(pub):>3} published {len(tracked_hits):>3} tracked {docs:>4} binary docs"
if not detail: return [line]
return [line] + [f"PUBLISHED {h}" for h in pub] + [f"tracked {h}" for h in tracked_hits]
def fleet_scan(
detail:bool = False, # List every finding, not just the per-submodule counts
strict:bool = False, # Exit 1 on any finding, not only on a published one
root:str = None, # Superproject root (default: nearest ancestor with .gitmodules)
):
"Find personal data and credentials in tracked files, published ones first"
style = "block" if detail else "line"
rep = walk(lambda p, s: scan_one(p, s, detail=detail), root=root,
workers=8, style=style)
code = finish(rep, "with findings")
if code: sys.exit(code)
# A published hit is live on the web and always fails. A tracked-but-unpublished
# one is a cleanup job, so it only fails under --strict.
published = sum(int(o.detail[0].split()[0]) for o in rep.ran if o.detail)
sys.exit(1 if published or (strict and rep.ran) else 0)scan_one
def scan_one(
path, sub:NoneType=None, detail:bool=False, nbs:str='nbs'
):Secret and personal-data findings for one submodule.
Only TRACKED files are scanned. An untracked file is on this disk and nowhere else, which is exactly where .gitignore is supposed to put a payslip, so flagging it would punish the fix.
Binary documents are counted rather than read. The count is informational and never fails the scan: a tracked PDF may be a payslip or may be a datasheet, and this cannot tell without a parser it has no business importing.
The fleet entry point
fleet
def fleet():Dispatch fleet <verb> to the matching command.
upload and status are already console scripts bound to the single-repo versions in automate, so the fleet-wide ones cannot claim those names. One script with verbs sidesteps the collision and keeps --help working per verb.
Exported source
VERBS = {"upload": fleet_upload, "push": fleet_push, "copy": fleet_copy,
"status": fleet_status, "update": fleet_update, "sync": fleet_sync,
"scan": fleet_scan}
def parse_verb(argv):
"(verb, remaining args) for `fleet ...`, or None when `argv` does not name a verb"
if not argv or argv[0].startswith("-"): return None
return (argv[0], argv[1:]) if argv[0] in VERBS else None
def usage():
"The verb list, for --help and for an unknown verb"
lines = ["Usage: fleet <verb> [options]", "", "Verbs:"]
lines += [f" {name:<8} {(fn.__doc__ or '').strip()}" for name, fn in sorted(VERBS.items())]
lines += ["", "Any verb takes --help for its own options."]
return "\n".join(lines)
def run_verb(name, argv):
"""Parse `argv` against the verb's own annotations, then call it.
Deliberately NOT `@call_parse`: fastcore runs the first decorated function in a module
the moment that module is `__main__`, which is exactly what makes `python -m
nbdevAuto.github` work with its single verb. With several verbs it would run whichever
happened to be defined first, so the verbs stay plain functions and the parser is built
here. `anno_parser` is the same one `call_parse` uses, so `--help` is unchanged."""
fn = VERBS[name]
args = vars(anno_parser(fn, prog=f"fleet {name}").parse_args(argv))
for k in ("pdb", "xtra"): args.pop(k, None)
return fn(**args)
def fleet():
"""Dispatch `fleet <verb>` to the matching command.
`upload` and `status` are already console scripts bound to the single-repo versions in
`automate`, so the fleet-wide ones cannot claim those names. One script with verbs
sidesteps the collision and keeps `--help` working per verb."""
picked = parse_verb(sys.argv[1:])
if not picked:
print(usage())
sys.exit(0 if sys.argv[1:2] in ([], ["-h"], ["--help"]) else 2)
return run_verb(*picked)run_verb
def run_verb(
name, argv
):Parse argv against the verb’s own annotations, then call it.
Deliberately NOT @call_parse: fastcore runs the first decorated function in a module the moment that module is __main__, which is exactly what makes python -m nbdevAuto.github work with its single verb. With several verbs it would run whichever happened to be defined first, so the verbs stay plain functions and the parser is built here. anno_parser is the same one call_parse uses, so --help is unchanged.
parse_verb
def parse_verb(
argv
):(verb, remaining args) for fleet ..., or None when argv does not name a verb
python -m nbdevAuto.fleet is the other way in, and the one the justfile falls back to when the installed nbdevAuto predates this module. #|export puts the guard in the library; #|eval: false keeps nbdev_test from running the CLI while testing the notebook.
Tests
A throwaway superproject of real git repos, rebuilt per run.
from fastcore.test import *
import tempfile
def _fixture(specs=(("a_dirty", True, True), ("b_clean", True, False), ("c_nonbs", False, False))):
"A throwaway superproject: (name, has nbs/, is dirty) per submodule"
d = Path(tempfile.mkdtemp())
lines = []
for name, nbs, dirty in specs:
p = d/name; p.mkdir()
subprocess.run(["git", "init", "-q", "-b", "main", str(p)], check=True)
for k, v in (("user.email", "t@example.com"), ("user.name", "t")):
subprocess.run(["git", "-C", str(p), "config", k, v], check=True)
(p/"README.md").write_text("x")
if nbs: (p/"nbs").mkdir()
subprocess.run(["git", "-C", str(p), "add", "."], check=True)
subprocess.run(["git", "-C", str(p), "commit", "-q", "-m", "init"], check=True)
if dirty: (p/"README.md").write_text("changed")
lines.append(f'[submodule "{name}"]\n\tpath = {name}\n\turl = git@github.com:me/{name}.git\n')
(d/".gitmodules").write_text("".join(lines))
return d
def _with_origin(p, extra=1):
"Give the repo at `p` a real bare origin, then leave it `extra` commits ahead of it"
bare = Path(f"{p}.origin.git")
subprocess.run(["git", "init", "-q", "--bare", str(bare)], check=True)
subprocess.run(["git", "-C", str(p), "remote", "add", "origin", str(bare)], check=True)
subprocess.run(["git", "-C", str(p), "push", "-q", "-u", "origin", "main"], check=True)
for i in range(extra):
(p/f"extra{i}.txt").write_text("x")
subprocess.run(["git", "-C", str(p), "add", f"extra{i}.txt"], check=True)
subprocess.run(["git", "-C", str(p), "commit", "-q", "-m", f"extra{i}"], check=True)
return bare
_root = _fixture()
def scan_file_text(text):
"Run the content scanners over a string, so a test need not make a file"
names = _hits(text)
addrs = {m.group(0).lower() for m in EMAIL.finditer(text)
if not RESERVED_DOMAINS.search(m.group(0))}
if len(addrs) >= BULK_EMAIL_MIN: names.append(f"{len(addrs)} email addresses")
return names# Only the dirty submodule with an nbs/ folder runs; the other two are counted, not narrated.
seen = []
rep = walk(lambda p, s: seen.append(s.path), root=_root, need_nbs=True, only_dirty=True, style="line")
test_eq(seen, ["a_dirty"])
test_eq(len(rep.ran), 1)
test_eq(rep.skip_counts(), {"without nbs": 1, "clean": 1})
test_eq(format_summary(rep, "uploaded"), "3 submodules: 1 uploaded, 1 clean, 1 without nbs")
test_eq(rep.code, 0)
# Dropping the dirty filter picks up both nbs/ submodules.
rep = walk(lambda p, s: None, root=_root, need_nbs=True, style="line")
test_eq([o.path for o in rep.ran], ["a_dirty", "b_clean"])
test_eq(format_summary(rep, "uploaded"), "3 submodules: 2 uploaded, 1 without nbs")
# A failure is named and counted, does not abort the others, and sets the exit code.
def _boom(p, s):
if s.path == "a_dirty": raise StepFailed("upload", 2, "fatal: nope")
return ["fine"]
rep = walk(_boom, root=_root, need_nbs=True, style="line")
test_eq([o.path for o in rep.failed], ["a_dirty"])
# A StepFailed's own message is the headline, because "upload exited 2" says nothing
test_eq(rep.failed[0].reason, "fatal: nope")
test_eq([o.path for o in rep.ran], ["b_clean"])
test_eq(rep.code, 1)
test_eq(format_summary(rep, "uploaded"), "3 submodules: 1 uploaded, 1 without nbs, 1 FAILED")
# With no output to quote, the command and exit code are the fallback headline.
def _silent(p, s): raise StepFailed("upload", 2)
test_eq(walk(_silent, root=_root, need_nbs=True, style="line").failed[0].reason, "upload exited 2")
# An uninitialised submodule is a directory with no .git, and is reported rather than run.
_bare = _fixture()
import shutil as _sh; _sh.rmtree(_bare/"b_clean"/".git")
test_eq(walk(lambda p, s: None, root=_bare, style="line").skip_counts(), {"not initialised": 1})
# Parallel and sequential runs agree, and both report in submodule order.
rep_p = walk(lambda p, s: [s.path], root=_root, workers=4, style="line")
rep_s = walk(lambda p, s: [s.path], root=_root, workers=1, style="line")
test_eq([o.path for o in rep_p.outcomes], [o.path for o in rep_s.outcomes])
test_eq([o.detail for o in rep_p.outcomes], [o.detail for o in rep_s.outcomes])a_dirty
a_dirty
b_clean
a_dirty fatal: nope
b_clean fine
a_dirty upload exited 2
b_clean upload exited 2
a_dirty
c_nonbs
a_dirty a_dirty
b_clean b_clean
c_nonbs c_nonbs
a_dirty a_dirty
b_clean b_clean
c_nonbs c_nonbs# Verb dispatch: a verb, its args, and everything that is not one.
test_eq(parse_verb(["upload", "--all"]), ("upload", ["--all"]))
test_eq(parse_verb(["upload"]), ("upload", []))
test_eq(parse_verb([]), None)
test_eq(parse_verb(["--help"]), None)
test_eq(parse_verb(["nosuchverb"]), None)
assert "upload" in usage() and "Usage: fleet" in usage()
# `run_verb` parses against the verb's own signature. Checked on a throwaway verb so the
# test does not have to run a real upload.
def _echo(all:bool=False, root:str=None):
"Echo verb, for tests"
return (all, root)
VERBS["_echo"] = _echo
try:
test_eq(run_verb("_echo", []), (False, None))
test_eq(run_verb("_echo", ["--all"]), (True, None))
test_eq(run_verb("_echo", ["--root", "/tmp/x"]), (False, "/tmp/x"))
test_eq(parse_verb(["_echo", "--all"]), ("_echo", ["--all"]))
finally:
del VERBS["_echo"]
# `script_path` says what to do rather than failing obscurely.
try:
script_path("definitely-not-a-console-script-xyz")
assert False, "a missing script must raise"
except RuntimeError as e:
assert "uv sync" in str(e)Phase 2: push and copy
# --- push -------------------------------------------------------------------------
_r2 = _fixture()
_with_origin(_r2/"b_clean", extra=2)
test_eq(ahead_of_remote(_r2/"b_clean"), (2, "main"))
test_eq(push_one(_r2/"b_clean", dry_run=True), ["would push 2 commits on main"])
# dry_run must not move the remote
test_eq(ahead_of_remote(_r2/"b_clean"), (2, "main"))
push_one(_r2/"b_clean")
test_eq(ahead_of_remote(_r2/"b_clean"), (0, "main"))
# The three reasons a submodule is passed over, each named rather than silently ignored.
def _skip_reason(fn, *a, **kw):
try: fn(*a, **kw); return None
except Skip as e: return str(e)
test_eq(_skip_reason(push_one, _r2/"b_clean"), "nothing to push")
test_eq(_skip_reason(push_one, _r2/"a_dirty"), "no remote")
subprocess.run(["git", "-C", str(_r2/"c_nonbs"), "checkout", "-q", "--detach"], check=True)
test_eq(_skip_reason(push_one, _r2/"c_nonbs"), "detached HEAD")
# A runtime Skip is counted, not printed, and never a failure.
rep = walk(lambda p, s: push_one(p, s, dry_run=True), root=_r2, style="line")
test_eq(rep.ran, [])
test_eq(rep.skip_counts(), {"no remote": 1, "nothing to push": 1, "detached HEAD": 1})
test_eq(rep.code, 0)
# --- copy -------------------------------------------------------------------------
_r3 = _fixture()
_shared = _r3/"shared.yml"; _shared.write_text("a: 1")
test_eq(copy_one(_r3/"a_dirty", files=[_shared], dry_run=True), ["would copy shared.yml -> nbs/"])
assert not (_r3/"a_dirty"/"nbs"/"shared.yml").exists(), "dry_run wrote a file"
test_eq(copy_one(_r3/"a_dirty", files=[_shared]), ["copied shared.yml -> nbs/"])
test_eq((_r3/"a_dirty"/"nbs"/"shared.yml").read_text(), "a: 1")
# No destination directory is a skip; a missing SOURCE is a failure, not a quiet no-op.
test_eq(_skip_reason(copy_one, _r3/"c_nonbs", files=[_shared]), "no nbs/")
try:
copy_one(_r3/"a_dirty", files=["/nope/missing.yml"])
assert False, "a missing source must fail"
except StepFailed as e:
assert "no such file" in e.output
rep = walk(lambda p, s: copy_one(p, s, files=[_shared]), root=_r3, workers=4, style="line")
test_eq(len(rep.ran), 2)
test_eq(format_summary(rep, "updated"), "3 submodules: 2 updated, 1 no nbs/")Phases 4 and 5: status, update, sync
# --- status -----------------------------------------------------------------------
_r4 = _fixture()
_line = status_one(_r4/"a_dirty")[0]
assert _line.startswith("main") and "dirty" in _line
assert "clean" in status_one(_r4/"b_clean")[0]
# --short lists the changed paths, and passes over the clean ones
_out = status_one(_r4/"a_dirty", short=True)
assert any("README.md" in l for l in _out[1:]), _out
# The FIRST porcelain line must not be shifted: `git status -s` pads the code to two
# columns and a strip would eat that leading space on line one only.
test_eq([l[:5] for l in _out[1:]], [" M "] * len(_out[1:]))
test_eq(_skip_reason(status_one, _r4/"b_clean", short=True), "clean")
rep = walk(lambda p, s: status_one(p, s, short=True), root=_r4, style="line")
test_eq(format_summary(rep, "changed"), "3 submodules: 1 changed, 2 clean")
# A detached HEAD is named, not shown as a branch called HEAD
subprocess.run(["git", "-C", str(_r4/"c_nonbs"), "checkout", "-q", "--detach"], check=True)
assert status_one(_r4/"c_nonbs")[0].startswith("DETACHED")
# --- update -----------------------------------------------------------------------
_r5 = _fixture()
_p = _r5/"b_clean"
_with_origin(_p, extra=2)
subprocess.run(["git", "-C", str(_p), "push", "-q"], check=True) # remote now level
subprocess.run(["git", "-C", str(_p), "reset", "-q", "--hard", "HEAD~2"], check=True)
test_eq(_count(_p, "HEAD..origin/main"), 2) # 2 behind
test_eq(update_one(_p, dry_run=True), ["would fast-forward 2"])
test_eq(_count(_p, "HEAD..origin/main"), 2) # dry run moved nothing
test_eq(update_one(_p), ["fast-forwarded 2 commits"])
test_eq(_count(_p, "HEAD..origin/main"), 0)
test_eq(_skip_reason(update_one, _p), "up to date")
# Diverged is a FAILURE, not a quiet skip: it needs a human and the exit code must say so.
subprocess.run(["git", "-C", str(_p), "reset", "-q", "--hard", "HEAD~1"], check=True)
(_p/"mine.txt").write_text("mine")
subprocess.run(["git", "-C", str(_p), "add", "mine.txt"], check=True)
subprocess.run(["git", "-C", str(_p), "commit", "-q", "-m", "mine"], check=True)
try:
update_one(_p)
assert False, "a diverged submodule must fail"
except StepFailed as e:
assert "diverged" in e.output and "resolve by hand" in e.output
# A dirty worktree is refused before anything is checked out.
try:
update_one(_r5/"a_dirty")
assert False, "a dirty submodule must fail"
except StepFailed as e:
assert "uncommitted changes" in e.output
test_eq(_skip_reason(update_one, _r5/"c_nonbs"), "no remote")
# A detached HEAD is reattached to main.
_q = _fixture()/"a_dirty"
subprocess.run(["git", "-C", str(_q), "checkout", "-q", "--", "README.md"], check=True)
_with_origin(_q, extra=0)
subprocess.run(["git", "-C", str(_q), "checkout", "-q", "--detach"], check=True)
test_eq(update_one(_q, dry_run=True), ["would checkout main (from DETACHED)"])
test_eq(update_one(_q), ["checked out main (was DETACHED)"])
test_eq(git(_q, "rev-parse", "--abbrev-ref", "HEAD"), "main")
# --- sync helpers -----------------------------------------------------------------
test_eq(dirty_submodules(_r4), ["a_dirty"])
# init_new only ever touches submodules git itself calls uninitialised
test_eq(init_new(_r4, dry_run=True), [])The scan verb
# --- looks_secret: the filter that decides whether the scan is worth running ----
# Every one of these false positives is a REAL line from the fleet's published
# teaching notebooks. They are the regression test, not hypotheticals.
for expr in ["forms.CharField(widget=forms.PasswordInput)", # Django forms page
"settings.STRIPE_SECRET_KEY", # Stripe page
"cleaned_data.get(", # Django forms page
"&&(i[0]+=t._mapboxAccessToken", # minified JS in a Plotly output
"ghp_************************************", # already masked
"YOUR_API_KEY", "changeme", "short"]:
test_eq((expr, looks_secret(expr)), (expr, False))
# A generated secret mixes case and digits over a decent length; a word does not.
for secret in ["b4bTbQgLdRCvoBZQSjaRlUh8lnslIflVDFet6VeToen4GWu-MYwHilUddyWZorKPFCSFiu1KL", # scan:allow
"EAAYp9C2YZAYYBO93tDbfKZAa5vp3ZB0fCh8yZCVv6VuWof1GxfZArv83OUlWoGI4jRYn1T6"]: # scan:allow
test_eq(looks_secret(secret), True)
# --- is_published mirrors the project.render exclude list in _quarto.yml -------
test_eq(is_published(Path("nbs/01_Tax.ipynb")), True)
test_eq(is_published(Path("nbs/index.ipynb")), True)
test_eq(is_published(Path("nbs/p_04_Pay.ipynb")), False) # excluded by project.render
test_eq(is_published(Path("nbs/data.csv")), False) # not a rendered suffix
# --- scan_file reads a notebook's OUTPUTS, not just its source ----------------
# The leak that prompted issue #18: the code cell was rewritten, the stale output
# holding the old figures was not, and Quarto publishes the output verbatim.
_d = Path(tempfile.mkdtemp())
_leaky = {"cells": [{"cell_type": "code", "source": ["print(summary)\n"], "outputs": [ # scan:allow
{"output_type": "stream", "text": ["token = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'\n"]}]}]} # scan:allow
(_d/"01_stale.ipynb").write_text(json.dumps(_leaky))
test_eq(scan_file(_d/"01_stale.ipynb"), ["credential"])
# An image payload is skipped: megabytes of base64 manufacture entropy hits.
_plot = {"cells": [{"cell_type": "code", "source": [], "outputs": [
{"output_type": "display_data", "data": {"image/png": "iVBORw0KGgoAAAANSUhEUg" * 500}}]}]}
(_d/"02_plot.ipynb").write_text(json.dumps(_plot))
test_eq(scan_file(_d/"02_plot.ipynb"), [])
# A column of addresses is someone's user table; one address is a contact detail.
# A real domain: the reserved ones are excluded, which the next cell tests.
(_d/"users.csv").write_text("email\n" + "".join(f"u{i}@clinic.com.au\n" for i in range(9)))
test_eq(scan_file(_d/"users.csv"), ["9 email addresses"])
(_d/"one.csv").write_text("owner\nben@clinic.com.au\n")
test_eq(scan_file(_d/"one.csv"), [])
# A binary document is never read, so it can never be a content finding.
(_d/"payslip.pdf").write_bytes(b"%PDF-1.4 token = aB3dEfGh1jKlMn0pQrStUvWxYz456789") # scan:allow
test_eq(scan_file(_d/"payslip.pdf"), [])
# --- the three filters that stopped the fleet's own docs from reporting -------
# RFC 2606 addresses are a mail tutorial, not a contact list.
test_eq(scan_file_text("".join(f"u{i}@example.com\n" for i in range(9))), [])
# A placeholder hash is hash-shaped but carries no material.
test_eq(scan_file_text("password: argon2:$argon2id$v=19$m=10240,t=10,p=8$<salt>$<hash>"), [])
test_eq(scan_file_text("password: argon2:$argon2id$v=19$m=1$75Q7uCcjzIK2HqrZ$" # scan:allow
"y8eYdSDFeAEAZW0S74AqnSMq3CAIl4hSBpdgkSVg"), ["password hash"])
# The armour without key material is every VPN page in the fleet.
test_eq(scan_file_text("-----BEGIN PRIVATE KEY-----\n(Insert Client Key)"), [])
# An inline marker is how this notebook stops flagging its own test vectors.
test_eq(scan_file_text("api_key = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'"), ["credential"]) # scan:allow
test_eq(scan_file_text("api_key = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789' # scan:allow"), [])# --- scan_one: only TRACKED files count, and only submodules with nbs/ ---------
_r6 = _fixture(specs=(("leaky", True, False), ("clean", True, False), ("nonbs", False, False)))
_leak = _r6/"leaky"/"nbs"
(_leak/"01_public.ipynb").write_text(json.dumps( # scan:allow
{"cells": [{"cell_type": "code", "source": ["api_key = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'\n"], # scan:allow
"outputs": []}]}))
(_leak/"p_02_private.ipynb").write_text(json.dumps(
{"cells": [{"cell_type": "code", "source": ["api_key = 'zY9xWv8uTs7rQp6oNm5lKj4iHg3fEd2c'\n"], # scan:allow
"outputs": []}]}))
(_leak/"receipts.pdf").write_bytes(b"%PDF-1.4")
subprocess.run(["git", "-C", str(_r6/"leaky"), "add", "-A"], check=True)
subprocess.run(["git", "-C", str(_r6/"leaky"), "commit", "-q", "-m", "data"], check=True)
# Untracked: on this disk and nowhere else, which is where .gitignore is meant to
# put a payslip. Flagging it would punish the fix.
(_leak/"99_untracked.ipynb").write_text(json.dumps(
{"cells": [{"cell_type": "code", "source": ["api_key = 'Qq1Ww2Ee3Rr4Tt5Yy6Uu7Ii8Oo9Pp0Aa'\n"], # scan:allow
"outputs": []}]}))
_out = scan_one(_r6/"leaky", detail=True)
test_eq(_out[0], f"{1:>3} published {1:>3} tracked {1:>4} binary docs")
test_eq(_out[1], "PUBLISHED nbs/01_public.ipynb: credential")
test_eq(_out[2], "tracked nbs/p_02_private.ipynb: credential")
test_eq(len(_out), 3) # the untracked file is absent
# A submodule with nothing to report skips, so the fleet line never appears for it.
test_fail(lambda: scan_one(_r6/"clean"), contains="clean")
test_fail(lambda: scan_one(_r6/"nonbs"), contains="without nbs")
# The walk reports only the submodule that found something; the rest are counted.
_rep6 = walk(lambda p, s: scan_one(p, s), root=_r6, style="line")
test_eq([o.path for o in _rep6.ran], ["leaky"])
test_eq(_rep6.skip_counts(), {"clean": 1, "without nbs": 1})
test_eq(format_summary(_rep6, "with findings"),
"3 submodules: 1 with findings, 1 clean, 1 without nbs")