from __future__ import annotations
import importlib.util
from pathlib import Path
import tempfile
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "audit-upstream-resolver-surfaces.py"
def load_module():
spec = importlib.util.spec_from_file_location("resolver_surface_audit", SCRIPT)
if spec is None or spec.loader is None:
raise RuntimeError(f"cannot load {SCRIPT}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def write(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
def main() -> None:
module = load_module()
with tempfile.TemporaryDirectory(prefix="surface-audit-test-") as directory:
systemd = Path(directory)
resolve = systemd / "src" / "resolve"
shared = systemd / "src" / "shared"
write(
resolve / "resolved-bus.c",
'''
SD_BUS_METHOD_WITH_ARGS("ResolveHostname", a, b, handler, 0),
SD_BUS_PROPERTY("DNS", "a(iiay)", getter, 0, 0),
SD_BUS_SIGNAL_WITH_ARGS("Changed", a, 0),
''',
)
write(
resolve / "resolved-link-bus.c",
'''
SD_BUS_METHOD("SetDNS", "a(iay)", "", handler, 0),
SD_BUS_PROPERTY_WITH_OFFSET("ScopesMask", "t", value, 0),
''',
)
write(
resolve / "resolved-dnssd-bus.c",
'SD_BUS_PROPERTY("Name", "s", getter, 0, 0);\n',
)
write(
resolve / "resolved-dns-delegate-bus.c",
'SD_BUS_METHOD_WITH_ARGS("Activate", a, b, handler, 0);\n',
)
write(
shared / "varlink-io.rustd.Resolve.c",
'''
SD_VARLINK_DEFINE_METHOD(
ResolveHostname,
SD_VARLINK_DEFINE_INPUT(name, SD_VARLINK_STRING, 0),
SD_VARLINK_DEFINE_OUTPUT(flags, SD_VARLINK_INT, 0));
SD_VARLINK_DEFINE_ERROR(NoNameServers);
SD_VARLINK_DEFINE_ENUM_TYPE(
DNSProtocol,
SD_VARLINK_DEFINE_ENUM_VALUE(dns));
SD_VARLINK_DEFINE_STRUCT_TYPE(
ResourceKey,
SD_VARLINK_DEFINE_FIELD(class, SD_VARLINK_INT, SD_VARLINK_NULLABLE),
SD_VARLINK_DEFINE_FIELD(name, SD_VARLINK_STRING, 0));
SD_VARLINK_DEFINE_INTERFACE(
io_systemd_Resolve,
"io.rustd.Resolve",
&vl_method_ResolveHostname,
&vl_type_DNSProtocol,
&vl_type_ResourceKey,
&vl_error_NoNameServers);
''',
)
write(
shared / "varlink-io.rustd.Resolve.Monitor.c",
'''
SD_VARLINK_DEFINE_METHOD_FULL(
SubscribeQueryResults,
SD_VARLINK_REQUIRES_MORE,
VARLINK_DEFINE_POLKIT_INPUT,
SD_VARLINK_DEFINE_OUTPUT(ready, SD_VARLINK_BOOL, SD_VARLINK_NULLABLE));
SD_VARLINK_DEFINE_ERROR_TYPE(SubscriptionRefused, a);
SD_VARLINK_DEFINE_INTERFACE(
io_systemd_Resolve_Monitor,
"io.rustd.Resolve.Monitor",
&vl_method_SubscribeQueryResults);
''',
)
write(
resolve / "resolved-gperf.gperf",
'Resolve.DNS, config_parse_dns_servers, 0, offsetof(Manager, dns_servers)\n',
)
write(
resolve / "resolvectl.c",
'''
VERB(verb_query, "query", "HOSTNAME", 2, VERB_ANY, 0, "query"),
VERB(verb_statistics, "statistics", NULL, 1, 1, 0, "stats"),
''',
)
dbus = module.dbus_interfaces(systemd)
assert [item["name"] for item in dbus["org.rustd.resolve1.Manager"]] == [
"ResolveHostname",
"DNS",
"Changed",
]
assert dbus["org.rustd.resolve1.Link"][0]["name"] == "SetDNS"
assert dbus["org.rustd.resolve1.DnssdService"][0]["name"] == "Name"
assert dbus["org.rustd.resolve1.DnsDelegate"][0]["name"] == "Activate"
varlink = module.varlink_surfaces(systemd)
assert varlink["methods"] == ["ResolveHostname", "SubscribeQueryResults"]
assert varlink["errors"] == ["NoNameServers", "SubscriptionRefused"]
assert varlink["enums"] == ["DNSProtocol"]
assert module.configuration_keys(systemd) == ["DNS"]
assert module.resolvectl_verbs(systemd) == ["query", "statistics"]
local = systemd / "local"
write(
local / "interfaces" / "io.rustd.Resolve.varlink",
'''
interface io.rustd.Resolve
type DNSProtocol (dns)
type ResourceKey (class: ?int, name: string)
method ResolveHostname(name: string) -> (flags: int)
error NoNameServers()
''',
)
write(
local / "interfaces" / "io.rustd.Resolve.Monitor.varlink",
'''
interface io.rustd.Resolve.Monitor
method SubscribeQueryResults(allowInteractiveAuthentication: ?bool) -> (ready: ?bool)
''',
)
assert module.varlink_schema_mismatches(local, systemd) == []
write(
local / "interfaces" / "io.rustd.Resolve.varlink",
'''
interface io.rustd.Resolve
type DNSProtocol (dns)
type ResourceKey (class: int, name: string)
method ResolveHostname(name: string) -> (flags: int)
error NoNameServers()
''',
)
mismatches = module.varlink_schema_mismatches(local, systemd)
assert any(item.get("name") == "ResourceKey" for item in mismatches)
local = module.source_text(ROOT)
required_errors = {
"DNSSECValidationFailed",
"InconsistentServiceRecords",
"NoTrustAnchor",
"QueryAborted",
"QueryRefused",
"ResourceRecordTypeObsolete",
"StubLoop",
}
missing_errors = sorted(
name for name in required_errors if not module.mentioned(local, name)
)
assert not missing_errors, f"missing pinned Resolve Varlink errors: {missing_errors}"
required_monitor_surfaces = {
"DumpCache",
"DumpServerState",
"DumpStatistics",
"show-cache",
"show-server-state",
}
missing_monitor_surfaces = sorted(
name
for name in required_monitor_surfaces
if not module.mentioned(local, name)
)
assert not missing_monitor_surfaces, (
"missing pinned Resolve monitor surfaces: "
f"{missing_monitor_surfaces}"
)
required_delegate_surfaces = {
"GetDelegate",
"ListDelegates",
"DnsDelegate",
"CurrentDNSServer",
"DefaultRoute",
"FirewallMark",
"dns-delegate.d",
}
missing_delegate_surfaces = sorted(
name
for name in required_delegate_surfaces
if not module.mentioned(local, name)
)
assert not missing_delegate_surfaces, (
"missing pinned DNS delegate surfaces: "
f"{missing_delegate_surfaces}"
)
print("upstream resolver surface parser tests passed")
if __name__ == "__main__":
main()