from __future__ import annotations
import argparse
import json
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
from typing import Any
class AuditError(RuntimeError):
pass
def command(*arguments: str, cwd: Path | None = None) -> str:
try:
return subprocess.check_output(
list(arguments),
cwd=cwd,
text=True,
stderr=subprocess.PIPE,
).strip()
except subprocess.CalledProcessError as error:
detail = error.stderr.strip() if error.stderr else str(error)
raise AuditError(f"command failed: {' '.join(arguments)}: {detail}") from error
def snake_case(value: str) -> str:
first = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", value)
return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", first).lower()
def source_text(root: Path) -> str:
parts = []
for directory in (root / "src", root / "ffi", root / "nss"):
if not directory.is_dir():
continue
for path in sorted(directory.rglob("*")):
if path.suffix not in {".rs", ".c", ".h", ".f90"} or not path.is_file():
continue
parts.append(path.read_text(encoding="utf-8", errors="replace"))
return "\n".join(parts)
def dbus_interfaces(systemd: Path) -> dict[str, list[dict[str, Any]]]:
sources = {
"org.freedesktop.resolve1.Manager": "resolved-bus.c",
"org.freedesktop.resolve1.Link": "resolved-link-bus.c",
"org.freedesktop.resolve1.DnssdService": "resolved-dnssd-bus.c",
"org.freedesktop.resolve1.DnsDelegate": "resolved-dns-delegate-bus.c",
}
patterns = {
"method": re.compile(
r"SD_BUS_METHOD(?:_[A-Z_]+)?\(\s*\"([^\"]+)\""
),
"property": re.compile(
r"SD_BUS_PROPERTY(?:_[A-Z_]+)?\(\s*\"([^\"]+)\""
),
"signal": re.compile(
r"SD_BUS_SIGNAL(?:_[A-Z_]+)?\(\s*\"([^\"]+)\""
),
}
output: dict[str, list[dict[str, Any]]] = {}
resolve = systemd / "src" / "resolve"
for interface, filename in sources.items():
path = resolve / filename
if not path.is_file():
raise AuditError(f"missing pinned D-Bus source: {path}")
text = path.read_text(encoding="utf-8", errors="replace")
members = []
for kind, pattern in patterns.items():
for name in sorted(set(pattern.findall(text))):
members.append(
{
"name": name,
"kind": kind,
"source": path.relative_to(systemd).as_posix(),
}
)
members.sort(key=lambda item: (item["kind"], item["name"]))
if members:
output[interface] = members
if not output:
raise AuditError("pinned D-Bus inventory is empty")
return dict(sorted(output.items()))
def varlink_surfaces(systemd: Path) -> dict[str, list[str]]:
methods: set[str] = set()
errors: set[str] = set()
enums: set[str] = set()
paths = sorted((systemd / "src" / "shared").glob("varlink-io.systemd.Resolve*.c"))
paths.extend(sorted((systemd / "src" / "resolve").glob("varlink-*.c")))
if not paths:
raise AuditError("pinned Varlink sources are missing")
for path in paths:
text = path.read_text(encoding="utf-8", errors="replace")
methods.update(
re.findall(
r"SD_VARLINK_DEFINE_METHOD(?:_[A-Z_]+)?\(\s*([A-Za-z0-9_]+)",
text,
)
)
errors.update(
re.findall(
r"SD_VARLINK_DEFINE_ERROR(?:_[A-Z_]+)?\(\s*([A-Za-z0-9_.]+)",
text,
)
)
enums.update(
re.findall(
r"SD_VARLINK_DEFINE_ENUM_TYPE(?:_[A-Z_]+)?\(\s*([A-Za-z0-9_]+)",
text,
)
)
output = {
"methods": sorted(methods),
"errors": sorted(errors),
"enums": sorted(enums),
}
if not output["methods"]:
raise AuditError("pinned Varlink method inventory is empty")
return output
def c_without_comments(text: str) -> str:
text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
return re.sub(r"//[^\n]*", "", text)
def macro_bodies(text: str, name: str) -> list[str]:
pattern = re.compile(rf"\b{re.escape(name)}(?:_[A-Z_]+)?\s*\(")
output = []
for match in pattern.finditer(text):
depth = 1
cursor = match.end()
quote: str | None = None
escaped = False
while cursor < len(text) and depth:
character = text[cursor]
if quote:
if escaped:
escaped = False
elif character == "\\":
escaped = True
elif character == quote:
quote = None
elif character in {'"', "'"}:
quote = character
elif character == "(":
depth += 1
elif character == ")":
depth -= 1
cursor += 1
if depth:
raise AuditError(f"unterminated {name} macro")
output.append(text[match.end() : cursor - 1])
return output
def macro_arguments(body: str) -> list[str]:
arguments = []
start = 0
depth = 0
quote: str | None = None
escaped = False
for index, character in enumerate(body):
if quote:
if escaped:
escaped = False
elif character == "\\":
escaped = True
elif character == quote:
quote = None
continue
if character in {'"', "'"}:
quote = character
elif character == "(":
depth += 1
elif character == ")":
depth -= 1
elif character == "," and depth == 0:
arguments.append(body[start:index].strip())
start = index + 1
arguments.append(body[start:].strip())
return arguments
def c_varlink_field_type(type_name: str, flags: str, named: bool) -> str:
if named:
value = type_name
else:
value = {
"SD_VARLINK_BOOL": "bool",
"SD_VARLINK_INT": "int",
"SD_VARLINK_STRING": "string",
"SD_VARLINK_OBJECT": "object",
"SD_VARLINK_FLOAT": "float",
}.get(type_name)
if value is None:
raise AuditError(f"unknown Varlink field type: {type_name}")
if "SD_VARLINK_ARRAY" in flags:
value = f"[]{value}"
if "SD_VARLINK_NULLABLE" in flags:
value = f"?{value}"
return value
def c_varlink_fields(body: str, field_kind: str) -> list[list[str]]:
fields = []
macro = f"SD_VARLINK_DEFINE_{field_kind}"
for match in re.finditer(rf"\b{macro}(?:_BY_TYPE)?\s*\(", body):
nested = macro_bodies(body[match.start() :], macro)[0]
arguments = macro_arguments(nested)
if len(arguments) < 3:
raise AuditError(f"malformed {macro} declaration")
named = body[match.start() : match.end()].split("(", 1)[0].endswith(
"_BY_TYPE"
)
fields.append(
[
arguments[0],
c_varlink_field_type(arguments[1], arguments[2], named),
]
)
return fields
def upstream_varlink_schema(systemd: Path) -> dict[str, dict[str, Any]]:
shared = systemd / "src" / "shared"
paths = [
shared / "varlink-io.systemd.Resolve.c",
shared / "varlink-io.systemd.Resolve.Monitor.c",
]
if not all(path.is_file() for path in paths):
raise AuditError("pinned Resolve Varlink schema sources are missing")
text = c_without_comments(
"\n".join(path.read_text(encoding="utf-8", errors="replace") for path in paths)
)
types: dict[str, Any] = {}
for body in macro_bodies(text, "SD_VARLINK_DEFINE_ENUM_TYPE"):
arguments = macro_arguments(body)
name = arguments[0]
values = re.findall(r"SD_VARLINK_DEFINE_ENUM_VALUE\(\s*([A-Za-z0-9_]+)", body)
types[name] = {"kind": "enum", "values": values}
for body in macro_bodies(text, "SD_VARLINK_DEFINE_STRUCT_TYPE"):
name = macro_arguments(body)[0]
types[name] = {
"kind": "struct",
"fields": c_varlink_fields(body, "FIELD"),
}
methods: dict[str, Any] = {}
for body in macro_bodies(text, "SD_VARLINK_DEFINE_METHOD"):
name = macro_arguments(body)[0]
inputs = c_varlink_fields(body, "INPUT")
if "VARLINK_DEFINE_POLKIT_INPUT" in body and not any(
field[0] == "allowInteractiveAuthentication" for field in inputs
):
inputs.append(["allowInteractiveAuthentication", "?bool"])
methods[name] = {
"inputs": inputs,
"outputs": c_varlink_fields(body, "OUTPUT"),
}
errors: dict[str, Any] = {}
for body in macro_bodies(text, "SD_VARLINK_DEFINE_ERROR"):
name = macro_arguments(body)[0]
errors[name] = {"fields": c_varlink_fields(body, "FIELD")}
interfaces: dict[str, dict[str, Any]] = {}
for body in macro_bodies(text, "SD_VARLINK_DEFINE_INTERFACE"):
arguments = macro_arguments(body)
symbol = arguments[0]
interface = {
"io_systemd_Resolve": "io.systemd.Resolve",
"io_systemd_Resolve_Monitor": "io.systemd.Resolve.Monitor",
}.get(symbol)
if interface is None:
continue
selected_types = re.findall(r"&vl_type_([A-Za-z0-9_]+)", body)
selected_methods = re.findall(r"&vl_method_([A-Za-z0-9_]+)", body)
selected_errors = re.findall(r"&vl_error_([A-Za-z0-9_]+)", body)
interfaces[interface] = {
"types": {name: types[name] for name in selected_types},
"methods": {name: methods[name] for name in selected_methods},
"errors": {name: errors[name] for name in selected_errors},
}
if set(interfaces) != {"io.systemd.Resolve", "io.systemd.Resolve.Monitor"}:
raise AuditError("pinned Resolve Varlink interface declarations are incomplete")
return interfaces
def idl_fields(body: str) -> list[list[str]]:
body = body.strip()
if not body:
return []
fields = []
for field in macro_arguments(body):
if ":" not in field:
raise AuditError(f"malformed Varlink IDL field: {field}")
name, type_name = field.split(":", 1)
fields.append([name.strip(), re.sub(r"\s+", "", type_name)])
return fields
def local_varlink_schema(root: Path) -> dict[str, dict[str, Any]]:
interfaces = {}
directory = root / "interfaces"
for path in sorted(directory.glob("io.systemd.Resolve*.varlink")):
text = path.read_text(encoding="utf-8")
first = re.search(
r"^\s*interface\s+([A-Za-z0-9_.]+)\s*$", text, re.MULTILINE
)
if first is None:
raise AuditError(f"missing Varlink interface declaration: {path}")
schema: dict[str, dict[str, Any]] = {
"types": {},
"methods": {},
"errors": {},
}
for line in text.splitlines():
line = line.strip()
if not line or line.startswith("interface "):
continue
type_match = re.fullmatch(r"type\s+([A-Za-z0-9_]+)\s*\((.*)\)", line)
if type_match:
name, body = type_match.groups()
if ":" in body:
schema["types"][name] = {
"kind": "struct",
"fields": idl_fields(body),
}
else:
schema["types"][name] = {
"kind": "enum",
"values": [value.strip() for value in body.split(",")],
}
continue
method_match = re.fullmatch(
r"method\s+([A-Za-z0-9_]+)\s*\((.*)\)\s*->\s*\((.*)\)",
line,
)
if method_match:
name, inputs, outputs = method_match.groups()
schema["methods"][name] = {
"inputs": idl_fields(inputs),
"outputs": idl_fields(outputs),
}
continue
error_match = re.fullmatch(r"error\s+([A-Za-z0-9_]+)\s*\((.*)\)", line)
if error_match:
name, fields = error_match.groups()
schema["errors"][name] = {"fields": idl_fields(fields)}
continue
raise AuditError(f"unsupported Varlink IDL declaration in {path}: {line}")
interfaces[first.group(1)] = schema
return interfaces
def varlink_schema_mismatches(root: Path, systemd: Path) -> list[dict[str, Any]]:
upstream = upstream_varlink_schema(systemd)
local = local_varlink_schema(root)
mismatches = []
for interface, expected in upstream.items():
actual = local.get(interface)
if actual is None:
mismatches.append({"interface": interface, "category": "interface"})
continue
for category in ("types", "methods", "errors"):
for name, signature in expected[category].items():
if actual[category].get(name) != signature:
mismatches.append(
{
"interface": interface,
"category": category,
"name": name,
"expected": signature,
"actual": actual[category].get(name),
}
)
for name in sorted(set(actual[category]) - set(expected[category])):
mismatches.append(
{
"interface": interface,
"category": category,
"name": name,
"expected": None,
"actual": actual[category][name],
}
)
return mismatches
def configuration_keys(systemd: Path) -> list[str]:
keys: set[str] = set()
for path in sorted((systemd / "src" / "resolve").glob("*gperf*")):
text = path.read_text(encoding="utf-8", errors="replace")
keys.update(re.findall(r"\bResolve\.([A-Za-z0-9]+)\b", text))
if not keys:
raise AuditError("pinned resolved.conf inventory is empty")
return sorted(keys)
def resolvectl_verbs(systemd: Path) -> list[str]:
path = systemd / "src" / "resolve" / "resolvectl.c"
if not path.is_file():
raise AuditError(f"missing pinned resolvectl source: {path}")
text = path.read_text(encoding="utf-8", errors="replace")
verbs = sorted(set(re.findall(r"\bVERB\(\s*[^,]+,\s*\"([^\"]+)\"", text)))
if not verbs:
raise AuditError("pinned resolvectl verb inventory is empty")
return verbs
def mentioned(text: str, name: str) -> bool:
candidates = {
name,
snake_case(name),
name.replace("-", "_"),
name.lower(),
}
return any(candidate and candidate in text for candidate in candidates)
def audit(root: Path, systemd: Path) -> dict[str, Any]:
local = source_text(root)
dbus = dbus_interfaces(systemd)
varlink = varlink_surfaces(systemd)
varlink_schema = varlink_schema_mismatches(root, systemd)
config = configuration_keys(systemd)
verbs = resolvectl_verbs(systemd)
missing_dbus = []
for interface, members in dbus.items():
for member in members:
if not mentioned(local, member["name"]):
missing_dbus.append(
{
"interface": interface,
"kind": member["kind"],
"name": member["name"],
}
)
missing_varlink_methods = [
name for name in varlink["methods"] if not mentioned(local, name)
]
missing_varlink_errors = [
name for name in varlink["errors"] if not mentioned(local, name)
]
missing_configuration = [name for name in config if not mentioned(local, name)]
missing_verbs = [name for name in verbs if not mentioned(local, name)]
suspicious = []
patterns = {
"todo_macro": re.compile(r"\btodo!\s*\("),
"unimplemented_macro": re.compile(r"\bunimplemented!\s*\("),
"not_supported_error": re.compile(r"NotSupported|not supported", re.I),
"placeholder_marker": re.compile(
r"\b(?:TODO|FIXME|placeholder)\b|implement(?:ed)? .* later|replace with real",
re.I,
),
}
for path in sorted((root / "src").rglob("*.rs")):
text = path.read_text(encoding="utf-8", errors="replace")
for category, pattern in patterns.items():
for match in pattern.finditer(text):
line = text.count("\n", 0, match.start()) + 1
suspicious.append(
{
"category": category,
"path": path.relative_to(root).as_posix(),
"line": line,
"excerpt": text[match.start() : match.start() + 100].splitlines()[0],
}
)
return {
"schema": 2,
"upstream_commit": command("git", "rev-parse", "HEAD", cwd=systemd),
"source_tree": command("git", "rev-parse", "HEAD^{tree}", cwd=root),
"dbus": dbus,
"varlink": varlink,
"configuration_keys": config,
"resolvectl_verbs": verbs,
"missing": {
"dbus": missing_dbus,
"varlink_methods": missing_varlink_methods,
"varlink_errors": missing_varlink_errors,
"varlink_schema": varlink_schema,
"configuration_keys": missing_configuration,
"resolvectl_verbs": missing_verbs,
},
"suspicious_implementation_markers": suspicious,
"counts": {
"dbus_members": sum(len(values) for values in dbus.values()),
"varlink_methods": len(varlink["methods"]),
"varlink_errors": len(varlink["errors"]),
"varlink_schema_mismatches": len(varlink_schema),
"configuration_keys": len(config),
"resolvectl_verbs": len(verbs),
"missing_total": (
len(missing_dbus)
+ len(missing_varlink_methods)
+ len(missing_varlink_errors)
+ len(varlink_schema)
+ len(missing_configuration)
+ len(missing_verbs)
),
},
}
def arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
parser.add_argument("--systemd-tree", type=Path)
parser.add_argument("--output", type=Path, default=Path("target/upstream-surface-audit.json"))
parser.add_argument("--fail-on-missing", action="store_true")
return parser.parse_args()
def main() -> int:
options = arguments()
root = options.root.resolve()
baseline = root / "compat" / "upstream-systemd"
commit = (baseline / "commit").read_text(encoding="ascii").strip()
temporary: Path | None = None
if options.systemd_tree:
systemd = options.systemd_tree.resolve()
else:
temporary = Path(tempfile.mkdtemp(prefix="resolved-surface-audit-"))
systemd = temporary / "systemd"
command(
"git",
"clone",
"--filter=blob:none",
"--no-checkout",
"https://github.com/systemd/systemd.git",
str(systemd),
)
command("git", "fetch", "--depth", "1", "origin", commit, cwd=systemd)
command("git", "checkout", "--detach", commit, cwd=systemd)
try:
if command("git", "rev-parse", "HEAD", cwd=systemd) != commit:
raise AuditError("systemd tree differs from the pinned commit")
report = audit(root, systemd)
output = options.output
if not output.is_absolute():
output = root / output
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
print(json.dumps(report["counts"], indent=2, sort_keys=True))
if options.fail_on_missing and report["counts"]["missing_total"]:
return 1
return 0
finally:
if temporary:
shutil.rmtree(temporary)
if __name__ == "__main__":
try:
raise SystemExit(main())
except (AuditError, OSError) as error:
print(f"audit-upstream-resolver-surfaces: {error}", file=sys.stderr)
raise SystemExit(2) from error