Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 86 additions & 14 deletions scripts/host_ids.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,8 +162,10 @@ def gather(repo: Path | None) -> Runtime:
rt.cpu_class = _cpu_class()
for f in ("product_name", "product_version", "product_family", "board_name"):
v = _read(f"/sys/class/dmi/id/{f}")
if len(v) >= 4 and not re.fullmatch(
r"(?i)none|default string|to be filled.*|\d+", v
if (
len(v) >= 4
and not re.fullmatch(r"(?i)none|default string|to be filled.*|\d+", v)
and not GENERIC_VM_PRODUCT.search(v)
):
rt.products.append(v)
if platform.system() == "Linux":
Expand Down Expand Up @@ -214,6 +216,11 @@ def _preempt(version: str) -> str:


INTEL_PART = r"i[3579]-\d{4,5}[A-Z]{0,2}"
# DMI product strings of hypervisors and cloud VMs: generic, never identifying.
GENERIC_VM_PRODUCT = re.compile(
r"(?i)virtual machine|standard pc|kvm|qemu|vmware|virtualbox|bochs|hvm domu|"
r"google compute engine|openstack|amazon ec2|droplet|cloud server"
)
KERNEL_BUILD = (
r"(?<!kernel-)(?<![\w.-])(\d+\.\d+)\.\d+-\d+(?:\.\w+)*?\.(?:fc|el)\d+(?:_\d+)?"
r"\.(?:x86_64|aarch64)(?!\.rpm)\b"
Expand Down Expand Up @@ -341,7 +348,17 @@ def runtime_rules(rt: Runtime) -> list[Rule]:
return rules


def on_ci_runner() -> bool:
"""True on a hosted CI runner, whose own values say nothing about the developer's machine."""
return os.environ.get("CI", "").lower() == "true" or "GITHUB_ACTIONS" in os.environ


def all_rules(rt: Runtime) -> list[Rule]:
"""This machine's literal values, then the generic patterns. On a CI runner only the
generic patterns apply: the runner's hostname, DMI and kernel are a cloud VM's, and
matching them would flag ordinary words ("Virtual Machine") in the repo."""
if on_ci_runner():
return generic_rules(rt.cpu_class)
return runtime_rules(rt) + generic_rules(rt.cpu_class)


Expand Down Expand Up @@ -469,17 +486,21 @@ def setup(repo_arg: str | None) -> tuple[Path, Config, list[Rule]]:

def cmd_check(repo: Path, cfg: Config, rules: list[Rule]) -> int:
n_files = n_hits = 0
literal = [r for r in rules if r.literal]
for rel in tracked(repo):
if cfg.skipped(rel) or is_self(repo, rel):
if cfg.skipped(rel):
continue
# The tool and its config carry fake examples of every pattern by design, so they
# are checked for this machine's own values only.
file_rules = literal if is_self(repo, rel) else rules
try:
text = decode((repo / rel).read_bytes())
except OSError:
continue
if text is None:
continue
n_files += 1
for line, cat in hits(text, rules, cfg, rel):
for line, cat in hits(text, file_rules, cfg, rel):
print(f"{rel}:{line}: {cat}")
n_hits += 1
print(f"host_ids check: {n_files} text file(s), {n_hits} hit(s)", file=sys.stderr)
Expand Down Expand Up @@ -565,10 +586,12 @@ def cmd_history(repo: Path, cfg: Config, rules: list[Rule], emit: str | None) ->
) # fmt: skip
assert batch.stdin and batch.stdout

def scan(text: str, rel: str) -> None:
def scan(text: str, rel: str, literal_only: bool = False) -> None:
for r in rules:
if cfg.allowed(r.category, rel) or cfg.skipped(rel):
continue
if literal_only and not r.literal:
continue
for m in r.pattern.finditer(text):
v = m.group(0)
if r.category == "home-path" and v in ("~", "<repo>"):
Expand All @@ -579,8 +602,9 @@ def scan(text: str, rel: str) -> None:
values.setdefault(v, _replacement_for(r, v))

for sha, rel in paths.items():
if Path(rel).name in (SELF.name, CONFIG_NAME):
continue
# Every past version of the tool is scanned for this machine's own values: a
# fixture once copied from a real file is exactly the leak to find.
own = Path(rel).name in (SELF.name, CONFIG_NAME)
batch.stdin.write(f"{sha}\n".encode())
batch.stdin.flush()
header = batch.stdout.readline().split()
Expand All @@ -592,7 +616,7 @@ def scan(text: str, rel: str) -> None:
continue
text = decode(data)
if text is not None:
scan(text, rel)
scan(text, rel, literal_only=own)
batch.stdin.close()
batch.wait()
messages = subprocess.run(
Expand All @@ -612,11 +636,9 @@ def scan(text: str, rel: str) -> None:
if not found:
print(" clean: nothing identifying in any reachable blob or message")
if emit:
lines = [
f"{v}==>{r}"
for v, r in sorted(values.items(), key=lambda kv: len(kv[0]), reverse=True)
if v != r
]
lines, dropped = format_rules(values)
if dropped:
print(f" {dropped} multi-line match(es) could not be written as rules")
Path(emit).write_text(
"\n".join(lines) + ("\n" if lines else ""), encoding="utf-8"
)
Expand All @@ -627,6 +649,28 @@ def scan(text: str, rel: str) -> None:
return 1 if found else 0


def format_rules(values: dict[str, str]) -> tuple[list[str], int]:
"""`git filter-repo --replace-text` lines, longest value first.

The format is one rule per line, so a match that spans a line break (a version on
the line after the word "kernel") is reduced to its last line when the replacement
keeps the same leading lines; otherwise it is dropped and counted. Every line is
prefixed `literal:` so a value can never be read as a regex or glob rule.
"""
rules: dict[str, str] = {}
dropped = 0
for v, r in values.items():
while "\n" in v and "\n" in r and v.split("\n", 1)[0] == r.split("\n", 1)[0]:
v, r = v.split("\n", 1)[1], r.split("\n", 1)[1]
if "\n" in v or "\n" in r or "==>" in v or not v.strip():
dropped += 1
continue
if v != r:
rules.setdefault(v, r)
ordered = sorted(rules.items(), key=lambda kv: len(kv[0]), reverse=True)
return [f"literal:{v}==>{r}" for v, r in ordered], dropped


def cmd_local(repo: Path) -> int:
g = ["git", "-C", str(repo)]
print("worktrees:")
Expand Down Expand Up @@ -705,7 +749,7 @@ def self_test() -> int:
)
subprocess.run(["git", "-C", str(repo), "add", "-A"], check=True)
fake.repo_path = str(repo)
rules = all_rules(fake)
rules = runtime_rules(fake) + generic_rules(fake.cpu_class)
cfg = load_config(repo)

def expect(cond: bool, what: str) -> None:
Expand Down Expand Up @@ -772,6 +816,34 @@ def expect(cond: bool, what: str) -> None:
expect(
cmd_scrub(repo, cfg, rules, [], dry_run=True) == 0, "scrub is idempotent"
)
lines, dropped = format_rules(
{
"kernel\n`8.4.2-200.fc98": "kernel\n`8.4",
"8.4.2-200.fc98.x86_64": "8.4",
"a\nb": "c\nd",
}
)
expect(
lines
== ["literal:8.4.2-200.fc98.x86_64==>8.4", "literal:`8.4.2-200.fc98==>`8.4"]
and dropped == 1,
f"format_rules: {lines!r} dropped={dropped}",
)
saved = {k: os.environ.get(k) for k in ("CI", "GITHUB_ACTIONS")}
os.environ["CI"] = "true"
expect(all(not r.literal for r in all_rules(fake)), "CI runner: generic rules only")
for k, v in saved.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
expect(
not GENERIC_VM_PRODUCT.search("Planted Laptop 9000"),
"laptop is not a VM product",
)
expect(
bool(GENERIC_VM_PRODUCT.search("Virtual Machine")), "Virtual Machine is generic"
)
# The tool must never carry this machine's own values, not even in a test fixture.
own = SELF.read_text(encoding="utf-8")
for r in runtime_rules(gather(None)):
Expand Down
Loading