Skip to content

watch

watch

shield watch — stream blocked-access events as JSON lines.

Tails the per-container audit log, (optionally) the NFLOG netlink socket, and the dnsmasq query log on the tiers that run dnsmasq. Clean exit on SIGINT or SIGTERM.

run_watch(state_dir, container)

Stream blocked-access events as JSON lines to stdout.

The audit log and the NFLOG socket feed every tier; the dnsmasq query log feeds only the tiers that run dnsmasq, so elsewhere the events carry IP addresses and no domain. Uses select so a single thread can multiplex the sources without blocking on any one of them.

Parameters:

Name Type Description Default
state_dir Path

Per-container state directory.

 required
container str

Container name (for event metadata).

 required

Raises:

Type Description
SystemExit

If the container recorded no DNS tier.
Source code in src/terok_shield/watch.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def run_watch(state_dir: Path, container: str) -> None:
    """Stream blocked-access events as JSON lines to stdout.

    The audit log and the NFLOG socket feed every tier; the dnsmasq query
    log feeds only the tiers that run dnsmasq, so elsewhere the events carry
    IP addresses and no domain.  Uses ``select`` so a single thread can
    multiplex the sources without blocking on any one of them.

    Args:
        state_dir: Per-container state directory.
        container: Container name (for event metadata).

    Raises:
        SystemExit: If the container recorded no DNS tier.
    """
    bundle = StateBundle(state_dir)
    tier = bundle.read_dns_tier()
    if tier is None:
        print("Error: DNS tier not set — container may not be shielded.", file=sys.stderr)
        raise SystemExit(1)

    _install_signal_handlers()

    dns_watcher = _dns_log_watcher(bundle, tier, container)
    audit_watcher = AuditLogWatcher(bundle.audit, container)
    nflog_watcher = NflogWatcher.create(container)
    domain_cache = DomainCache(state_dir)

    try:
        while _running:
            _poll_nflog_or_sleep(nflog_watcher, domain_cache)
            if dns_watcher:
                _emit_events(dns_watcher.poll())
            _emit_events(audit_watcher.poll())
    finally:
        if dns_watcher:
            dns_watcher.close()
        audit_watcher.close()
        if nflog_watcher:
            nflog_watcher.close()