From 16429c20c2ddeba464d214145dc051d4a9d44c18 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Wed, 1 Jul 2026 22:38:37 -0400 Subject: [PATCH] feat(zingela): M5 two-engine concurrency + truecolor dashboard + NDJSON Turn the scan from sequential TX-then-RX into two io.concurrent engines on std.Io.Threaded, fixing the M4 bug where replies arriving during transmit were dropped and the RX deadline (anchored at socket-open) expired before draining. The main thread is the non-blocking Io.Queue consumer and dashboard renderer. - rx: Receiver keyed off a shared tx_done atomic; new pure planDrain anchors the drain window at TX-completion, not socket-open, killing the quiet-gap early exit during a slow TX. Hard-cap safety backstop anchored at drain start. - output (new): cache-line-padded atomic Stats, truecolor/256/none palette (violet gradient, neon-green, muted-gray), in-place multi-line live dashboard, Unicode results table, NDJSON to stdout with visuals on stderr, full NO_COLOR/CLICOLOR_FORCE/COLORTERM/--color chain plus a narrow-terminal fallback. - tx: wall-clock deadline so a stalled TX ring can no longer hang the scan. - cli: banner recolored to the violet gradient and routed through the color chain. - targets: add 192.88.99.0/24 (6to4 relay anycast) to the RFC 6890 exclude floor. 78/78 unit tests pass under Debug and ReleaseSafe; proven end-to-end (open/closed, dedup, dashboard, NDJSON) in an unshare -r -n netns under ReleaseSafe. --- .../advanced/zig-stateless-scanner/build.zig | 13 +- .../zig-stateless-scanner/src/cli.zig | 24 +- .../zig-stateless-scanner/src/main.zig | 4 +- .../zig-stateless-scanner/src/netutil.zig | 14 + .../zig-stateless-scanner/src/output.zig | 578 ++++++++++++++++++ .../advanced/zig-stateless-scanner/src/rx.zig | 85 ++- .../zig-stateless-scanner/src/scancmd.zig | 231 +++++-- .../zig-stateless-scanner/src/targets.zig | 12 +- .../advanced/zig-stateless-scanner/src/tx.zig | 42 +- .../zig-stateless-scanner/src/txcmd.zig | 6 +- 10 files changed, 924 insertions(+), 85 deletions(-) create mode 100644 PROJECTS/advanced/zig-stateless-scanner/src/output.zig diff --git a/PROJECTS/advanced/zig-stateless-scanner/build.zig b/PROJECTS/advanced/zig-stateless-scanner/build.zig index e251b5ae..fdcc69a1 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/build.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/build.zig @@ -8,7 +8,7 @@ pub fn build(b: *std.Build) void { const optimize = b.standardOptimizeOption(.{}); const opts = b.addOptions(); - opts.addOption([]const u8, "version", "0.0.0-m4"); + opts.addOption([]const u8, "version", "0.0.0-m5"); const packet_mod = b.createModule(.{ .root_source_file = b.path("src/packet.zig"), @@ -89,6 +89,14 @@ pub fn build(b: *std.Build) void { classify_mod.addImport("packet", packet_mod); classify_mod.addImport("cookie", cookie_mod); + const output_mod = b.createModule(.{ + .root_source_file = b.path("src/output.zig"), + .target = target, + .optimize = optimize, + }); + output_mod.addImport("classify", classify_mod); + cli_mod.addImport("output", output_mod); + const dedup_mod = b.createModule(.{ .root_source_file = b.path("src/dedup.zig"), .target = target, @@ -137,6 +145,7 @@ pub fn build(b: *std.Build) void { scancmd_mod.addImport("rx", rx_mod); scancmd_mod.addImport("dedup", dedup_mod); scancmd_mod.addImport("netutil", netutil_mod); + scancmd_mod.addImport("output", output_mod); const exe = b.addExecutable(.{ .name = "zingela", @@ -167,7 +176,7 @@ pub fn build(b: *std.Build) void { smoke_step.dependOn(&smoke_cmd.step); const test_step = b.step("test", "Run unit tests"); - const test_mods = [_]*std.Build.Module{ packet_mod, cli_mod, smoke_mod, cookie_mod, numtheory_mod, targets_mod, ratelimit_mod, template_mod, afpacket_mod, tx_mod, txcmd_mod, classify_mod, dedup_mod, rx_mod, netutil_mod, scancmd_mod }; + const test_mods = [_]*std.Build.Module{ packet_mod, cli_mod, smoke_mod, cookie_mod, numtheory_mod, targets_mod, ratelimit_mod, template_mod, afpacket_mod, tx_mod, txcmd_mod, classify_mod, dedup_mod, rx_mod, netutil_mod, output_mod, scancmd_mod }; for (test_mods) |mod| { const t = b.addTest(.{ .root_module = mod }); const rt = b.addRunArtifact(t); diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig b/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig index 8a7d9ccd..37ed0c6f 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/cli.zig @@ -3,8 +3,7 @@ const std = @import("std"); const build_config = @import("build_config"); - -const reset = "\x1b[0m"; +const output = @import("output"); const banner_art = \\ ____ _ _ @@ -14,19 +13,12 @@ const banner_art = \\ |___/ ; -pub fn colorEnabled(io: std.Io) bool { - return std.Io.File.stdout().isTty(io) catch false; -} - -pub fn printBanner(io: std.Io) !void { - var buf: [512]u8 = undefined; +pub fn printBanner(io: std.Io, env: *std.process.Environ.Map) !void { + var buf: [1024]u8 = undefined; var fw = std.Io.File.stdout().writer(io, &buf); const out = &fw.interface; - if (colorEnabled(io)) { - try out.print("\x1b[38;2;000;200;255m{s}{s}\n", .{ banner_art, reset }); - } else { - try out.print("{s}\n", .{banner_art}); - } + const level = output.envLevel(io, std.Io.File.stdout(), env, .auto); + try output.bannerWordmark(out, level, banner_art); try out.print(" zingela {s} stateless mass scanner (Zig 0.16)\n\n", .{build_config.version}); try out.flush(); } @@ -39,8 +31,8 @@ pub fn printVersion(io: std.Io) !void { try out.flush(); } -pub fn printHelp(io: std.Io) !void { - try printBanner(io); +pub fn printHelp(io: std.Io, env: *std.process.Environ.Map) !void { + try printBanner(io, env); var buf: [512]u8 = undefined; var fw = std.Io.File.stdout().writer(io, &buf); const out = &fw.interface; @@ -67,6 +59,8 @@ pub fn printHelp(io: std.Io) !void { \\ \\scan-only options: \\ --wait receive drain window after transmit (default 2000) + \\ --json emit NDJSON results to stdout (visuals go to stderr) + \\ --color auto | always | never (default auto) \\ \\authorized use only. responsible default rate; needs CAP_NET_RAW \\(grant once: sudo setcap cap_net_raw,cap_net_admin=eip ./zig-out/bin/zingela) diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/main.zig b/PROJECTS/advanced/zig-stateless-scanner/src/main.zig index 2939aa6c..3b352b60 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/main.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/main.zig @@ -25,7 +25,7 @@ pub fn main(init: std.process.Init) !void { return txcmd.run(io, arena, args); } if (std.mem.eql(u8, cmd, "scan")) { - return scancmd.run(io, arena, args); + return scancmd.run(io, arena, args, init.environ_map); } - return cli.printHelp(io); + return cli.printHelp(io, init.environ_map); } diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/netutil.zig b/PROJECTS/advanced/zig-stateless-scanner/src/netutil.zig index 66a78ba2..adb5a6d5 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/netutil.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/netutil.zig @@ -12,6 +12,13 @@ pub fn getFlag(args: []const []const u8, name: []const u8) ?[]const u8 { return null; } +pub fn hasFlag(args: []const []const u8, name: []const u8) bool { + for (args) |a| { + if (std.mem.eql(u8, a, name)) return true; + } + return false; +} + pub fn parseIpv4(text: []const u8) !u32 { var addr: u32 = 0; var octets: usize = 0; @@ -116,3 +123,10 @@ test "getFlag finds values and tolerates missing" { try std.testing.expectEqualStrings("5000", getFlag(&args, "--rate").?); try std.testing.expect(getFlag(&args, "--target") == null); } + +test "hasFlag detects a valueless boolean flag in any position" { + const args = [_][]const u8{ "scan", "--target", "10.0.0.0/24", "--json" }; + try std.testing.expect(hasFlag(&args, "--json")); + try std.testing.expect(hasFlag(&args, "scan")); + try std.testing.expect(!hasFlag(&args, "--nope")); +} diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/output.zig b/PROJECTS/advanced/zig-stateless-scanner/src/output.zig new file mode 100644 index 00000000..4a9bc18c --- /dev/null +++ b/PROJECTS/advanced/zig-stateless-scanner/src/output.zig @@ -0,0 +1,578 @@ +// ©AngelaMos | 2026 +// output.zig + +const std = @import("std"); +const classify = @import("classify"); + +pub const State = classify.State; +pub const Result = classify.Result; + +const esc = "\x1b"; +const sgr_reset = "\x1b[0m"; +const clear_line = "\x1b[2K"; +const block_full = "\u{2588}"; +const block_light = "\u{2591}"; +const gutter_bar = "\u{258e}"; +const box_h = "\u{2500}"; + +const cache_line = std.atomic.cache_line; + +pub const Counter = std.atomic.Value(u64); + +pub const Padded = struct { + v: Counter align(cache_line) = .{ .raw = 0 }, + _pad: [cache_line - @sizeOf(Counter)]u8 = undefined, +}; + +pub const Stats = struct { + sent: Padded = .{}, + found: Padded = .{}, + open: Padded = .{}, + closed: Padded = .{}, + filtered: Padded = .{}, + + pub fn record(self: *Stats, st: State) void { + _ = self.found.v.fetchAdd(1, .monotonic); + switch (st) { + .open => _ = self.open.v.fetchAdd(1, .monotonic), + .closed => _ = self.closed.v.fetchAdd(1, .monotonic), + .filtered => _ = self.filtered.v.fetchAdd(1, .monotonic), + } + } +}; + +const Rgb = struct { r: u8, g: u8, b: u8 }; + +const violet_light = Rgb{ .r = 167, .g = 139, .b = 250 }; +const violet_mid = Rgb{ .r = 139, .g = 92, .b = 246 }; +const violet_deep = Rgb{ .r = 109, .g = 74, .b = 255 }; +const neon_green = Rgb{ .r = 74, .g = 222, .b = 128 }; +const chrome_gray = Rgb{ .r = 120, .g = 120, .b = 140 }; +const bright_white = Rgb{ .r = 230, .g = 230, .b = 240 }; +const soft_amber = Rgb{ .r = 217, .g = 164, .b = 74 }; + +pub const ColorLevel = enum { none, ansi256, truecolor }; +pub const ColorChoice = enum { auto, always, never }; + +pub fn parseColorChoice(text: ?[]const u8) ColorChoice { + const t = text orelse return .auto; + if (std.mem.eql(u8, t, "always")) return .always; + if (std.mem.eql(u8, t, "never")) return .never; + return .auto; +} + +pub fn resolveLevel(choice: ColorChoice, colored: bool, truecolor: bool) ColorLevel { + return switch (choice) { + .never => .none, + .always => if (truecolor) .truecolor else .ansi256, + .auto => if (!colored) .none else if (truecolor) .truecolor else .ansi256, + }; +} + +pub fn detectLevel( + io: std.Io, + file: std.Io.File, + choice: ColorChoice, + no_color: bool, + clicolor_force: bool, + truecolor: bool, +) ColorLevel { + switch (choice) { + .never => return .none, + .always => return resolveLevel(.always, true, truecolor), + .auto => { + const mode = std.Io.Terminal.Mode.detect(io, file, no_color, clicolor_force) catch return .none; + const colored = switch (mode) { + .no_color => false, + else => true, + }; + return resolveLevel(.auto, colored, truecolor); + }, + } +} + +pub fn envLevel(io: std.Io, file: std.Io.File, env: *std.process.Environ.Map, choice: ColorChoice) ColorLevel { + const no_color = if (env.get("NO_COLOR")) |v| v.len > 0 else false; + const clicolor_force = if (env.get("CLICOLOR_FORCE")) |v| (v.len > 0 and !std.mem.eql(u8, v, "0")) else false; + const colorterm = env.get("COLORTERM") orelse ""; + const truecolor = std.mem.eql(u8, colorterm, "truecolor") or std.mem.eql(u8, colorterm, "24bit"); + return detectLevel(io, file, choice, no_color, clicolor_force, truecolor); +} + +pub fn bannerWordmark(out: *std.Io.Writer, level: ColorLevel, art: []const u8) !void { + if (level == .none) { + try out.print("{s}\n", .{art}); + return; + } + const span_lines: f32 = 4.0; + var it = std.mem.splitScalar(u8, art, '\n'); + var i: usize = 0; + while (it.next()) |line| : (i += 1) { + const t = @min(@as(f32, @floatFromInt(i)), span_lines) / span_lines; + try setFg(out, level, violetAt(t)); + try out.print("{s}\n", .{line}); + } + try resetFg(out, level); +} + +fn to256(c: Rgb) u8 { + const r6: u16 = (@as(u16, c.r) * 5 + 127) / 255; + const g6: u16 = (@as(u16, c.g) * 5 + 127) / 255; + const b6: u16 = (@as(u16, c.b) * 5 + 127) / 255; + return @intCast(16 + 36 * r6 + 6 * g6 + b6); +} + +fn setFg(out: *std.Io.Writer, level: ColorLevel, c: Rgb) !void { + switch (level) { + .none => {}, + .truecolor => try out.print("\x1b[38;2;{d};{d};{d}m", .{ c.r, c.g, c.b }), + .ansi256 => try out.print("\x1b[38;5;{d}m", .{to256(c)}), + } +} + +fn resetFg(out: *std.Io.Writer, level: ColorLevel) !void { + if (level != .none) try out.writeAll(sgr_reset); +} + +fn span(out: *std.Io.Writer, level: ColorLevel, c: Rgb, text: []const u8) !void { + try setFg(out, level, c); + try out.writeAll(text); + try resetFg(out, level); +} + +fn lerpByte(a: u8, b: u8, t: f32) u8 { + const af: f32 = @floatFromInt(a); + const bf: f32 = @floatFromInt(b); + return @intFromFloat(std.math.clamp(af + (bf - af) * t, 0.0, 255.0)); +} + +fn lerp(a: Rgb, b: Rgb, t: f32) Rgb { + return .{ .r = lerpByte(a.r, b.r, t), .g = lerpByte(a.g, b.g, t), .b = lerpByte(a.b, b.b, t) }; +} + +fn violetAt(t: f32) Rgb { + const tc = std.math.clamp(t, 0.0, 1.0); + if (tc <= 0.5) return lerp(violet_light, violet_mid, tc * 2.0); + return lerp(violet_mid, violet_deep, (tc - 0.5) * 2.0); +} + +fn gradientText(out: *std.Io.Writer, level: ColorLevel, text: []const u8) !void { + if (level == .none) { + try out.writeAll(text); + return; + } + const n = text.len; + for (text, 0..) |ch, i| { + const t: f32 = if (n <= 1) 0.0 else @as(f32, @floatFromInt(i)) / @as(f32, @floatFromInt(n - 1)); + try setFg(out, level, violetAt(t)); + try out.writeByte(ch); + } + try resetFg(out, level); +} + +const bar_width: usize = 22; + +fn progressBar(out: *std.Io.Writer, level: ColorLevel, frac: f64) !void { + const clamped = std.math.clamp(frac, 0.0, 1.0); + const filled: usize = @intFromFloat(clamped * @as(f64, @floatFromInt(bar_width))); + var i: usize = 0; + while (i < bar_width) : (i += 1) { + if (i < filled) { + const t: f32 = if (bar_width <= 1) 0.0 else @as(f32, @floatFromInt(i)) / @as(f32, @floatFromInt(bar_width - 1)); + try setFg(out, level, violetAt(t)); + try out.writeAll(block_full); + } else { + try setFg(out, level, chrome_gray); + try out.writeAll(block_light); + } + } + try resetFg(out, level); +} + +fn writeThousands(out: *std.Io.Writer, n: u64) !void { + var buf: [24]u8 = undefined; + const s = std.fmt.bufPrint(&buf, "{d}", .{n}) catch return; + for (s, 0..) |ch, i| { + if (i != 0 and (s.len - i) % 3 == 0) try out.writeByte(','); + try out.writeByte(ch); + } +} + +fn writeClock(out: *std.Io.Writer, secs: u64) !void { + try out.print("{d:0>2}:{d:0>2}:{d:0>2}", .{ secs / 3600, (secs % 3600) / 60, secs % 60 }); +} + +fn writeIp(out: *std.Io.Writer, ip: u32) !void { + try out.print("{d}.{d}.{d}.{d}", .{ (ip >> 24) & 0xff, (ip >> 16) & 0xff, (ip >> 8) & 0xff, ip & 0xff }); +} + +fn stateName(st: State) []const u8 { + return switch (st) { + .open => "open", + .closed => "closed", + .filtered => "filtered", + }; +} + +fn stateLabel(st: State) []const u8 { + return switch (st) { + .open => "OPEN", + .closed => "CLOSED", + .filtered => "FILTERED", + }; +} + +fn stateColor(st: State) Rgb { + return switch (st) { + .open => neon_green, + .closed => chrome_gray, + .filtered => soft_amber, + }; +} + +pub const Dashboard = struct { + level: ColorLevel, + interactive: bool, + total: u64, + drawn: bool = false, + + const body_lines: usize = 5; + + pub fn init(level: ColorLevel, interactive: bool, total: u64) Dashboard { + return .{ .level = level, .interactive = interactive, .total = total }; + } + + fn gutter(self: *const Dashboard, out: *std.Io.Writer) !void { + try out.writeAll(" "); + try span(out, self.level, violet_mid, gutter_bar); + try out.writeByte(' '); + } + + pub fn render(self: *Dashboard, out: *std.Io.Writer, s: *const Stats, elapsed_ns: u64) !void { + const sent = s.sent.v.load(.monotonic); + const found = s.found.v.load(.monotonic); + const op = s.open.v.load(.monotonic); + const cl = s.closed.v.load(.monotonic); + const fi = s.filtered.v.load(.monotonic); + + const elapsed_s = @as(f64, @floatFromInt(elapsed_ns)) / 1_000_000_000.0; + const pps = if (elapsed_s > 0) @as(f64, @floatFromInt(sent)) / elapsed_s else 0; + const kpps = pps / 1000.0; + const frac = if (self.total > 0) @as(f64, @floatFromInt(sent)) / @as(f64, @floatFromInt(self.total)) else 0; + const pct = frac * 100.0; + const remaining = if (self.total > sent) self.total - sent else 0; + const eta_s: u64 = if (pps > 1.0) @intFromFloat(@as(f64, @floatFromInt(remaining)) / pps) else 0; + + if (!self.interactive) { + try out.print( + "[up {d:.0}s] sent {d} / {d} found {d} (open {d} closed {d} filtered {d}) {d:.2} kpps\n", + .{ elapsed_s, sent, self.total, found, op, cl, fi, kpps }, + ); + try out.flush(); + return; + } + + if (self.drawn) try out.print("\x1b[{d}A", .{body_lines}); + self.drawn = true; + + try out.writeAll(clear_line); + try out.writeAll(" "); + try span(out, self.level, violet_mid, gutter_bar); + try out.writeByte(' '); + try gradientText(out, self.level, "zingela"); + try span(out, self.level, chrome_gray, " scanning"); + try out.writeByte('\n'); + + try out.writeAll(clear_line); + try self.gutter(out); + try span(out, self.level, chrome_gray, "rate "); + try setFg(out, self.level, bright_white); + try out.print("{d:>8.2}", .{kpps}); + try span(out, self.level, chrome_gray, " kpps "); + try progressBar(out, self.level, frac); + try out.writeByte(' '); + try setFg(out, self.level, bright_white); + try out.print("{d:>5.1}", .{pct}); + try span(out, self.level, chrome_gray, "%"); + try resetFg(out, self.level); + try out.writeByte('\n'); + + try out.writeAll(clear_line); + try self.gutter(out); + try span(out, self.level, chrome_gray, "sent "); + try setFg(out, self.level, bright_white); + try writeThousands(out, sent); + try span(out, self.level, chrome_gray, " / "); + try setFg(out, self.level, chrome_gray); + try writeThousands(out, self.total); + try resetFg(out, self.level); + try out.writeByte('\n'); + + try out.writeAll(clear_line); + try self.gutter(out); + try span(out, self.level, chrome_gray, "open "); + try setFg(out, self.level, neon_green); + try writeThousands(out, op); + try span(out, self.level, chrome_gray, " closed "); + try setFg(out, self.level, chrome_gray); + try writeThousands(out, cl); + try span(out, self.level, chrome_gray, " filtered "); + try setFg(out, self.level, soft_amber); + try writeThousands(out, fi); + try resetFg(out, self.level); + try out.writeByte('\n'); + + try out.writeAll(clear_line); + try self.gutter(out); + try span(out, self.level, chrome_gray, "found "); + try setFg(out, self.level, neon_green); + try writeThousands(out, found); + try span(out, self.level, chrome_gray, " up "); + try setFg(out, self.level, bright_white); + try writeClock(out, @intFromFloat(elapsed_s)); + try span(out, self.level, chrome_gray, " eta "); + try setFg(out, self.level, bright_white); + try writeClock(out, eta_s); + try resetFg(out, self.level); + try out.writeByte('\n'); + + try out.flush(); + } +}; + +pub fn emitJson(out: *std.Io.Writer, r: Result) !void { + try out.print("{{\"ip\":\"", .{}); + try writeIp(out, r.ip); + try out.print("\",\"port\":{d},\"proto\":\"tcp\",\"state\":\"{s}\"}}\n", .{ r.port, stateName(r.state) }); +} + +const w_host: usize = 17; +const w_port: usize = 7; +const w_state: usize = 10; + +fn repeat(out: *std.Io.Writer, cell: []const u8, n: usize) !void { + var i: usize = 0; + while (i < n) : (i += 1) try out.writeAll(cell); +} + +fn rule(out: *std.Io.Writer, level: ColorLevel, left: []const u8, mid: []const u8, right: []const u8) !void { + try setFg(out, level, chrome_gray); + try out.writeAll(left); + try repeat(out, box_h, w_host + 2); + try out.writeAll(mid); + try repeat(out, box_h, w_port + 2); + try out.writeAll(mid); + try repeat(out, box_h, w_state + 2); + try out.writeAll(right); + try resetFg(out, level); + try out.writeByte('\n'); +} + +fn pad(out: *std.Io.Writer, n: usize) !void { + var i: usize = 0; + while (i < n) : (i += 1) try out.writeByte(' '); +} + +pub fn renderTable(out: *std.Io.Writer, level: ColorLevel, results: []const Result) !void { + try out.writeAll(" "); + try rule(out, level, "\u{250c}", "\u{252c}", "\u{2510}"); + + try out.writeAll(" "); + try span(out, level, chrome_gray, "\u{2502} "); + try setFg(out, level, bright_white); + try out.writeAll("HOST"); + try pad(out, w_host - "HOST".len); + try span(out, level, chrome_gray, " \u{2502} "); + try setFg(out, level, bright_white); + try pad(out, w_port - "PORT".len); + try out.writeAll("PORT"); + try span(out, level, chrome_gray, " \u{2502} "); + try setFg(out, level, bright_white); + try out.writeAll("STATE"); + try pad(out, w_state - "STATE".len); + try span(out, level, chrome_gray, " \u{2502}"); + try out.writeByte('\n'); + + try out.writeAll(" "); + try rule(out, level, "\u{251c}", "\u{253c}", "\u{2524}"); + + for (results) |r| { + var ipbuf: [15]u8 = undefined; + var ipw = std.Io.Writer.fixed(&ipbuf); + try writeIp(&ipw, r.ip); + const ip_str = ipbuf[0..ipw.end]; + + try out.writeAll(" "); + try span(out, level, chrome_gray, "\u{2502} "); + try setFg(out, level, bright_white); + try out.writeAll(ip_str); + try resetFg(out, level); + try pad(out, w_host - ip_str.len); + + try span(out, level, chrome_gray, " \u{2502} "); + var portbuf: [5]u8 = undefined; + const port_str = std.fmt.bufPrint(&portbuf, "{d}", .{r.port}) catch unreachable; + try pad(out, w_port - port_str.len); + try setFg(out, level, bright_white); + try out.writeAll(port_str); + try resetFg(out, level); + + try span(out, level, chrome_gray, " \u{2502} "); + const label = stateLabel(r.state); + try setFg(out, level, stateColor(r.state)); + try out.writeAll(label); + try resetFg(out, level); + try pad(out, w_state - label.len); + try span(out, level, chrome_gray, " \u{2502}"); + try out.writeByte('\n'); + } + + try out.writeAll(" "); + try rule(out, level, "\u{2514}", "\u{2534}", "\u{2518}"); +} + +pub fn ipPortLess(_: void, a: Result, b: Result) bool { + if (a.ip != b.ip) return a.ip < b.ip; + return a.port < b.port; +} + +pub fn renderSummary( + out: *std.Io.Writer, + level: ColorLevel, + sent: u64, + ifname: []const u8, + elapsed_s: f64, + open: u64, + closed: u64, + filtered: u64, +) !void { + try out.writeAll(" "); + try span(out, level, violet_mid, gutter_bar); + try out.writeByte(' '); + try span(out, level, chrome_gray, "sent "); + try setFg(out, level, bright_white); + try writeThousands(out, sent); + try span(out, level, chrome_gray, " SYN on "); + try setFg(out, level, bright_white); + try out.writeAll(ifname); + try span(out, level, chrome_gray, " in "); + try setFg(out, level, bright_white); + try out.print("{d:.3}s", .{elapsed_s}); + try span(out, level, chrome_gray, " \u{2192} "); + try setFg(out, level, neon_green); + try writeThousands(out, open); + try span(out, level, chrome_gray, " open "); + try setFg(out, level, chrome_gray); + try writeThousands(out, closed); + try span(out, level, chrome_gray, " closed "); + try setFg(out, level, soft_amber); + try writeThousands(out, filtered); + try span(out, level, chrome_gray, " filtered"); + try resetFg(out, level); + try out.writeByte('\n'); +} + +test "resolveLevel honors choice, tty state, and truecolor" { + try std.testing.expectEqual(ColorLevel.none, resolveLevel(.never, true, true)); + try std.testing.expectEqual(ColorLevel.truecolor, resolveLevel(.always, false, true)); + try std.testing.expectEqual(ColorLevel.ansi256, resolveLevel(.always, false, false)); + try std.testing.expectEqual(ColorLevel.none, resolveLevel(.auto, false, true)); + try std.testing.expectEqual(ColorLevel.truecolor, resolveLevel(.auto, true, true)); + try std.testing.expectEqual(ColorLevel.ansi256, resolveLevel(.auto, true, false)); +} + +test "parseColorChoice maps flag values" { + try std.testing.expectEqual(ColorChoice.auto, parseColorChoice(null)); + try std.testing.expectEqual(ColorChoice.always, parseColorChoice("always")); + try std.testing.expectEqual(ColorChoice.never, parseColorChoice("never")); + try std.testing.expectEqual(ColorChoice.auto, parseColorChoice("garbage")); +} + +test "to256 maps palette anchors into the 6x6x6 cube" { + try std.testing.expectEqual(@as(u8, 16), to256(.{ .r = 0, .g = 0, .b = 0 })); + try std.testing.expectEqual(@as(u8, 231), to256(.{ .r = 255, .g = 255, .b = 255 })); + try std.testing.expect(to256(neon_green) >= 16 and to256(neon_green) <= 231); +} + +test "violetAt interpolates the gradient endpoints and midpoint" { + try std.testing.expectEqual(violet_light, violetAt(0.0)); + try std.testing.expectEqual(violet_deep, violetAt(1.0)); + try std.testing.expectEqual(violet_mid, violetAt(0.5)); +} + +test "Padded counter is cache-line sized and aligned to prevent false sharing" { + try std.testing.expectEqual(@as(usize, cache_line), @sizeOf(Padded)); + try std.testing.expectEqual(@as(usize, cache_line), @alignOf(Padded)); +} + +test "Stats.record tallies per-state and total found without cross-talk" { + var s: Stats = .{}; + s.record(.open); + s.record(.open); + s.record(.closed); + s.record(.filtered); + try std.testing.expectEqual(@as(u64, 2), s.open.v.load(.monotonic)); + try std.testing.expectEqual(@as(u64, 1), s.closed.v.load(.monotonic)); + try std.testing.expectEqual(@as(u64, 1), s.filtered.v.load(.monotonic)); + try std.testing.expectEqual(@as(u64, 4), s.found.v.load(.monotonic)); + try std.testing.expectEqual(@as(u64, 0), s.sent.v.load(.monotonic)); +} + +test "emitJson writes one greppable NDJSON object per result" { + var buf: [128]u8 = undefined; + var w = std.Io.Writer.fixed(&buf); + try emitJson(&w, .{ .ip = 0x0a000005, .port = 80, .state = .open }); + try std.testing.expectEqualStrings( + "{\"ip\":\"10.0.0.5\",\"port\":80,\"proto\":\"tcp\",\"state\":\"open\"}\n", + buf[0..w.end], + ); +} + +test "writeThousands groups digits and leaves small numbers intact" { + var buf: [32]u8 = undefined; + var w = std.Io.Writer.fixed(&buf); + try writeThousands(&w, 1234567); + try std.testing.expectEqualStrings("1,234,567", buf[0..w.end]); + w = std.Io.Writer.fixed(&buf); + try writeThousands(&w, 42); + try std.testing.expectEqualStrings("42", buf[0..w.end]); +} + +test "renderTable with no color is plain and contains every host row" { + var buf: [1024]u8 = undefined; + var w = std.Io.Writer.fixed(&buf); + const rows = [_]Result{ + .{ .ip = 0x0a000005, .port = 80, .state = .open }, + .{ .ip = 0x0a000006, .port = 443, .state = .closed }, + }; + try renderTable(&w, .none, &rows); + const text = buf[0..w.end]; + try std.testing.expect(std.mem.indexOf(u8, text, "10.0.0.5") != null); + try std.testing.expect(std.mem.indexOf(u8, text, "OPEN") != null); + try std.testing.expect(std.mem.indexOf(u8, text, "10.0.0.6") != null); + try std.testing.expect(std.mem.indexOf(u8, text, "CLOSED") != null); + try std.testing.expect(std.mem.indexOf(u8, text, esc) == null); +} + +test "dashboard non-interactive frame is a single plain line" { + var buf: [256]u8 = undefined; + var w = std.Io.Writer.fixed(&buf); + var s: Stats = .{}; + _ = s.sent.v.fetchAdd(500, .monotonic); + s.record(.open); + var dash = Dashboard.init(.none, false, 1000); + try dash.render(&w, &s, 1_000_000_000); + const text = buf[0..w.end]; + try std.testing.expect(std.mem.indexOf(u8, text, "sent 500 / 1000") != null); + try std.testing.expect(std.mem.indexOf(u8, text, esc) == null); + try std.testing.expectEqual(@as(usize, 1), std.mem.count(u8, text, "\n")); +} + +test "ipPortLess orders by ip then port" { + const a = Result{ .ip = 0x0a000001, .port = 443, .state = .open }; + const b = Result{ .ip = 0x0a000001, .port = 80, .state = .open }; + const c = Result{ .ip = 0x0a000002, .port = 1, .state = .open }; + try std.testing.expect(ipPortLess({}, b, a)); + try std.testing.expect(ipPortLess({}, a, c)); + try std.testing.expect(!ipPortLess({}, a, b)); +} diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/rx.zig b/PROJECTS/advanced/zig-stateless-scanner/src/rx.zig index 134a90e3..abd7d993 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/rx.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/rx.zig @@ -45,10 +45,37 @@ pub const OpenError = error{ BindFailed, }; +const POLL_TICK_MS: i32 = 100; + +pub const RecvPlan = union(enum) { stop, poll: i32 }; + +pub fn planDrain( + now_ns: u64, + hard_deadline_ns: u64, + tx_done: bool, + drain_anchor_ns: *?u64, + drain_window_ns: u64, + tick_ms: i32, +) RecvPlan { + if (now_ns >= hard_deadline_ns) return .stop; + if (!tx_done) return .{ .poll = tick_ms }; + if (drain_anchor_ns.* == null) drain_anchor_ns.* = now_ns; + const deadline = drain_anchor_ns.*.? + drain_window_ns; + if (now_ns >= deadline) return .stop; + const remaining_ms: u64 = (deadline - now_ns) / NS_PER_MS; + const cap: u64 = @intCast(tick_ms); + const chosen: u64 = @min(remaining_ms + 1, cap); + return .{ .poll = @intCast(chosen) }; +} + pub const Receiver = struct { fd: i32, - quiet_ms: i32, - deadline_ns: u64, + tx_done: *std.atomic.Value(bool), + drain_window_ns: u64, + hard_cap_ns: u64, + started: bool = false, + hard_deadline_ns: u64 = 0, + drain_anchor_ns: ?u64 = null, fn monoNow() u64 { var ts: linux.timespec = undefined; @@ -56,7 +83,7 @@ pub const Receiver = struct { return @as(u64, @intCast(ts.sec)) * NS_PER_SEC + @as(u64, @intCast(ts.nsec)); } - pub fn open(ifname: []const u8, quiet_ms: i32, total_budget_ns: u64) OpenError!Receiver { + pub fn open(ifname: []const u8, tx_done: *std.atomic.Value(bool), drain_window_ns: u64, hard_cap_ns: u64) OpenError!Receiver { const rc_sock = linux.socket( linux.AF.PACKET, linux.SOCK.RAW, @@ -87,16 +114,21 @@ pub const Receiver = struct { if (linux.errno(linux.bind(fd, @ptrCast(&sll), @sizeOf(linux.sockaddr.ll))) != .SUCCESS) return error.BindFailed; - return .{ .fd = fd, .quiet_ms = quiet_ms, .deadline_ns = monoNow() + total_budget_ns }; + return .{ .fd = fd, .tx_done = tx_done, .drain_window_ns = drain_window_ns, .hard_cap_ns = hard_cap_ns }; } pub fn recv(self: *Receiver, buf: []u8) ?usize { - const quiet: u64 = if (self.quiet_ms > 0) @intCast(self.quiet_ms) else 0; while (true) { const now = monoNow(); - if (now >= self.deadline_ns) return null; - const remaining_ms = (self.deadline_ns - now) / NS_PER_MS; - const timeout: i32 = if (remaining_ms >= quiet) @intCast(quiet) else @intCast(remaining_ms); + if (!self.started) { + self.started = true; + self.hard_deadline_ns = now +| self.hard_cap_ns; + } + const done = self.tx_done.load(.acquire); + const timeout: i32 = switch (planDrain(now, self.hard_deadline_ns, done, &self.drain_anchor_ns, self.drain_window_ns, POLL_TICK_MS)) { + .stop => return null, + .poll => |t| t, + }; var pfd = [_]linux.pollfd{.{ .fd = self.fd, .events = linux.POLL.IN, .revents = 0 }}; const pr = linux.poll(&pfd, 1, timeout); switch (linux.errno(pr)) { @@ -104,7 +136,7 @@ pub const Receiver = struct { .INTR => continue, else => return null, } - if (pr == 0) return null; + if (pr == 0) continue; const rc = linux.recvfrom(self.fd, buf.ptr, buf.len, 0, null, null); switch (linux.errno(rc)) { .SUCCESS => return @intCast(rc), @@ -232,3 +264,38 @@ test "Io.Queue hands the deduped set from a producer to a consumer fiber" { try std.testing.expectEqual(@as(usize, 2), out.items.len); } + +const ms: u64 = 1_000_000; + +test "planDrain keeps polling while TX is still in flight (no quiet-gap early exit)" { + var anchor: ?u64 = null; + const p = planDrain(5 * ms, 100_000 * ms, false, &anchor, 2_000 * ms, POLL_TICK_MS); + switch (p) { + .poll => |t| try std.testing.expectEqual(POLL_TICK_MS, t), + .stop => return error.ShouldNotStopDuringTx, + } + try std.testing.expect(anchor == null); +} + +test "planDrain anchors the drain window at TX completion, not socket-open" { + var anchor: ?u64 = null; + const now: u64 = 30_000 * ms; + const p = planDrain(now, 100_000 * ms, true, &anchor, 2_000 * ms, POLL_TICK_MS); + try std.testing.expectEqual(@as(?u64, now), anchor); + switch (p) { + .poll => |t| try std.testing.expect(t >= 1 and t <= POLL_TICK_MS), + .stop => return error.ShouldStillDrain, + } +} + +test "planDrain stops once the post-TX drain window elapses" { + var anchor: ?u64 = 30_000 * ms; + const p = planDrain(32_001 * ms, 100_000 * ms, true, &anchor, 2_000 * ms, POLL_TICK_MS); + try std.testing.expect(std.meta.activeTag(p) == .stop); +} + +test "planDrain honors the hard safety cap even if TX never signalled done" { + var anchor: ?u64 = null; + const p = planDrain(9_999, 9_999, false, &anchor, 1, POLL_TICK_MS); + try std.testing.expect(std.meta.activeTag(p) == .stop); +} diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig b/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig index 02b044f0..3d576c63 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/scancmd.zig @@ -11,56 +11,146 @@ const tx = @import("tx"); const rx = @import("rx"); const dedup = @import("dedup"); const netutil = @import("netutil"); +const output = @import("output"); const default_iface = "lo"; const default_rate: u64 = 10_000; const default_src_port: u16 = 40_000; const default_wait_ms: i32 = 2_000; -const rx_max_drain_ms: i32 = 60_000; const ns_per_ms: u64 = 1_000_000; +const ns_per_sec: u64 = 1_000_000_000; const default_ports = [_]u16{80}; const dedup_capacity: usize = 1024; -const queue_capacity: usize = 256; +const queue_capacity: usize = 2048; +const drain_batch: usize = 256; +const drain_tick_ns: u64 = 50 * ns_per_ms; +const render_tick_interactive_ns: u64 = 125 * ns_per_ms; +const render_tick_plain_ns: u64 = 1_000 * ns_per_ms; +const rx_hard_cap_floor_ns: u64 = 60 * ns_per_sec; +const min_dashboard_cols: u16 = 64; const need_cap_hint = "scan: need CAP_NET_RAW + CAP_NET_ADMIN. Grant once, then re-run (no sudo):\n" ++ " sudo setcap cap_net_raw,cap_net_admin=eip ./zig-out/bin/zingela\nSkipping.\n"; -fn stateLabel(s: rx.State) []const u8 { - return switch (s) { - .open => "OPEN", - .closed => "CLOSED", - .filtered => "FILTERED", - }; +const concurrency_hint = + "scan: this system cannot launch concurrent TX/RX (needs >= 2 worker threads).\n"; + +const TxSink = struct { + backend: *afpacket.Backend, + sent: *output.Counter, + + pub fn submit(self: *TxSink, frame: []const u8) bool { + if (self.backend.submit(frame)) { + _ = self.sent.fetchAdd(1, .monotonic); + return true; + } + return false; + } + + pub fn kick(self: *TxSink) void { + self.backend.kick(); + } +}; + +fn txWorker( + engine: *targets.Engine, + tmpl: *const template.SynTemplate, + bucket: *ratelimit.TokenBucket, + sink: *TxSink, + max_packets: u64, + budget_ns: u64, + tx_done: *std.atomic.Value(bool), +) u64 { + var clock = netutil.RealClock{}; + const deadline_ns = clock.now() +| budget_ns; + const sent = tx.run(engine, tmpl, bucket, sink, &clock, max_packets, deadline_ns); + tx_done.store(true, .release); + return sent; } -fn consume(io: std.Io, queue: *std.Io.Queue(rx.Result), out: *std.ArrayList(rx.Result), allocator: std.mem.Allocator) void { - while (true) { - const r = queue.getOne(io) catch break; - out.append(allocator, r) catch {}; +fn rxWorker( + receiver: *rx.Receiver, + ck: cookie.Cookie, + dd: *dedup.Dedup, + sink: *rx.QueueSink, + rx_done: *std.atomic.Value(bool), +) void { + rx.run(receiver, ck, dd, sink); + rx_done.store(true, .release); +} + +fn absorb( + batch: []const rx.Result, + found: *std.ArrayList(rx.Result), + allocator: std.mem.Allocator, + stats: *output.Stats, + json_out: ?*std.Io.Writer, +) void { + for (batch) |r| { + found.append(allocator, r) catch continue; + stats.record(r.state); + if (json_out) |w| output.emitJson(w, r) catch {}; } } -pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) !void { - var buf: [1024]u8 = undefined; - var fw = std.Io.File.stdout().writer(io, &buf); - const out = &fw.interface; +fn drainQueue( + io: std.Io, + queue: *std.Io.Queue(rx.Result), + buf: []rx.Result, + found: *std.ArrayList(rx.Result), + allocator: std.mem.Allocator, + stats: *output.Stats, + json_out: ?*std.Io.Writer, +) void { + while (true) { + const n = queue.get(io, buf, 0) catch return; + if (n == 0) return; + absorb(buf[0..n], found, allocator, stats, json_out); + if (n < buf.len) return; + } +} + +fn terminalCols(fd: i32) ?u16 { + var ws: std.posix.winsize = undefined; + const rc = std.os.linux.ioctl(fd, std.os.linux.T.IOCGWINSZ, @intFromPtr(&ws)); + if (std.os.linux.errno(rc) != .SUCCESS) return null; + if (ws.col == 0) return null; + return ws.col; +} + +pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8, env: *std.process.Environ.Map) !void { + var obuf: [4096]u8 = undefined; + var ow = std.Io.File.stdout().writer(io, &obuf); + const out = &ow.interface; + + var ebuf: [4096]u8 = undefined; + var ew = std.Io.File.stderr().writer(io, &ebuf); + const derr = &ew.interface; const target_text = netutil.getFlag(args, "--target") orelse { - try out.writeAll("scan: --target is required (e.g. --target 10.0.0.0/24)\n"); - try out.flush(); + try derr.writeAll("scan: --target is required (e.g. --target 10.0.0.0/24)\n"); + try derr.flush(); return; }; const ifname = netutil.getFlag(args, "--iface") orelse default_iface; const rate = if (netutil.getFlag(args, "--rate")) |r| try std.fmt.parseInt(u64, r, 10) else default_rate; const src_port = if (netutil.getFlag(args, "--src-port")) |p| try std.fmt.parseInt(u16, p, 10) else default_src_port; const wait_ms = if (netutil.getFlag(args, "--wait")) |w| try std.fmt.parseInt(i32, w, 10) else default_wait_ms; + const json = netutil.hasFlag(args, "--json"); const ports = if (netutil.getFlag(args, "--ports")) |p| try netutil.parsePorts(allocator, p) else try allocator.dupe(u16, &default_ports); const gw_mac = if (netutil.getFlag(args, "--gw-mac")) |m| try netutil.parseMac(m) else [_]u8{0} ** 6; const src_ip = if (netutil.getFlag(args, "--src-ip")) |s| try netutil.parseIpv4(s) else try netutil.resolveSrcIp(ifname); const src_mac = try netutil.resolveSrcMac(ifname); + const choice = output.parseColorChoice(netutil.getFlag(args, "--color")); + const out_level = output.envLevel(io, std.Io.File.stdout(), env, choice); + const err_level = output.envLevel(io, std.Io.File.stderr(), env, choice); + const stderr_tty = std.Io.File.stderr().isTty(io) catch false; + const wide_enough = if (terminalCols(2)) |c| c >= min_dashboard_cols else true; + const interactive = stderr_tty and wide_enough; + var seed: u64 = undefined; if (netutil.getFlag(args, "--seed")) |s| { seed = try std.fmt.parseInt(u64, s, 10); @@ -74,6 +164,7 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) ! var eng = try targets.Engine.init(allocator, &.{cidr}, ports, seed); defer eng.deinit(); const count = if (netutil.getFlag(args, "--count")) |c| try std.fmt.parseInt(u64, c, 10) else eng.total; + const dash_total = @min(count, eng.total); const ck = try cookie.Cookie.random(io); const tmpl = template.SynTemplate.init(.{ @@ -87,19 +178,26 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) ! var backend = afpacket.Backend.open(ifname, .{}) catch |err| switch (err) { error.NeedCapNetRaw => { - try out.writeAll(need_cap_hint); - try out.flush(); + try derr.writeAll(need_cap_hint); + try derr.flush(); return; }, else => return err, }; defer backend.close(); - const rx_budget_ns: u64 = @as(u64, @intCast(@max(wait_ms, rx_max_drain_ms))) * ns_per_ms; - var receiver = rx.Receiver.open(ifname, wait_ms, rx_budget_ns) catch |err| switch (err) { + var tx_done = std.atomic.Value(bool).init(false); + var rx_done = std.atomic.Value(bool).init(false); + + const drain_window_ns: u64 = @as(u64, @intCast(@max(wait_ms, 0))) * ns_per_ms; + const est_tx_ns: u64 = if (rate > 0) (count / rate) *| ns_per_sec else rx_hard_cap_floor_ns; + const tx_budget_ns: u64 = (est_tx_ns *| 4) +| rx_hard_cap_floor_ns; + const hard_cap_ns: u64 = tx_budget_ns +| drain_window_ns; + + var receiver = rx.Receiver.open(ifname, &tx_done, drain_window_ns, hard_cap_ns) catch |err| switch (err) { error.NeedCapNetRaw => { - try out.writeAll(need_cap_hint); - try out.flush(); + try derr.writeAll(need_cap_hint); + try derr.flush(); return; }, else => return err, @@ -117,39 +215,76 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) ! const found_alloc = found_arena.allocator(); var found: std.ArrayList(rx.Result) = .empty; + var stats: output.Stats = .{}; + const json_out: ?*std.Io.Writer = if (json) out else null; + + try derr.print("zingela target {s} iface {s} rate {d} pps ports {d}\n", .{ target_text, ifname, rate, ports.len }); + try derr.flush(); + var clock = netutil.RealClock{}; const t0 = clock.now(); - var consumer = io.async(consume, .{ io, &queue, &found, found_alloc }); + var tx_sink = TxSink{ .backend = &backend, .sent = &stats.sent.v }; + var rx_sink = rx.QueueSink{ .queue = &queue, .io = io }; - const sent = tx.run(&eng, &tmpl, &bucket, &backend, &clock, count); + var tx_fut = io.concurrent(txWorker, .{ &eng, &tmpl, &bucket, &tx_sink, count, tx_budget_ns, &tx_done }) catch { + try derr.writeAll(concurrency_hint); + try derr.flush(); + return; + }; + var rx_fut = io.concurrent(rxWorker, .{ &receiver, ck, &dd, &rx_sink, &rx_done }) catch { + _ = tx_fut.await(io); + try derr.writeAll(concurrency_hint); + try derr.flush(); + return; + }; - var sink = rx.QueueSink{ .queue = &queue, .io = io }; - rx.run(&receiver, ck, &dd, &sink); + var dash = output.Dashboard.init(err_level, interactive, dash_total); + const render_interval_ns: u64 = if (interactive) render_tick_interactive_ns else render_tick_plain_ns; + var drain_buf: [drain_batch]rx.Result = undefined; + var last_render: u64 = 0; + while (!(tx_done.load(.acquire) and rx_done.load(.acquire))) { + drainQueue(io, &queue, drain_buf[0..], &found, found_alloc, &stats, json_out); + if (json) out.flush() catch {}; + const now = clock.now(); + if (last_render == 0 or now -| last_render >= render_interval_ns) { + dash.render(derr, &stats, now -| t0) catch {}; + last_render = now; + } + clock.sleepNs(drain_tick_ns); + } + + const sent = tx_fut.await(io); + rx_fut.await(io); queue.close(io); - consumer.await(io); + drainQueue(io, &queue, drain_buf[0..], &found, found_alloc, &stats, json_out); + if (json) out.flush() catch {}; - const elapsed_ns = clock.now() - t0; - const elapsed_s = @as(f64, @floatFromInt(elapsed_ns)) / 1_000_000_000.0; + dash.render(derr, &stats, clock.now() -| t0) catch {}; - var open_n: usize = 0; - var closed_n: usize = 0; - var filtered_n: usize = 0; - for (found.items) |r| { - const ip = r.ip; - try out.print("{d}.{d}.{d}.{d}:{d} {s}\n", .{ - (ip >> 24) & 0xff, (ip >> 16) & 0xff, (ip >> 8) & 0xff, ip & 0xff, r.port, stateLabel(r.state), - }); - switch (r.state) { - .open => open_n += 1, - .closed => closed_n += 1, - .filtered => filtered_n += 1, + var open_n: u64 = 0; + var closed_n: u64 = 0; + var filtered_n: u64 = 0; + for (found.items) |r| switch (r.state) { + .open => open_n += 1, + .closed => closed_n += 1, + .filtered => filtered_n += 1, + }; + + if (!json) { + if (found.items.len > 0) { + std.mem.sort(rx.Result, found.items, {}, output.ipPortLess); + try out.writeByte('\n'); + try output.renderTable(out, out_level, found.items); + try out.flush(); + } else { + try derr.writeAll(" no open, closed, or filtered responses observed\n"); } } - try out.print( - "scan: sent {d} SYN on {s} in {d:.3}s; {d} open, {d} closed, {d} filtered ({d} replies)\n", - .{ sent, ifname, elapsed_s, open_n, closed_n, filtered_n, found.items.len }, - ); - try out.flush(); + + const elapsed_s = @as(f64, @floatFromInt(clock.now() - t0)) / @as(f64, @floatFromInt(ns_per_sec)); + try derr.writeByte('\n'); + try output.renderSummary(derr, err_level, sent, ifname, elapsed_s, open_n, closed_n, filtered_n); + try derr.flush(); } diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/targets.zig b/PROJECTS/advanced/zig-stateless-scanner/src/targets.zig index 50de00e2..ab32a14a 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/targets.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/targets.zig @@ -22,6 +22,7 @@ const reserved = [_]Range{ .{ .start = 0xac100000, .end = 0xac1fffff }, .{ .start = 0xc0000000, .end = 0xc00000ff }, .{ .start = 0xc0000200, .end = 0xc00002ff }, + .{ .start = 0xc0586300, .end = 0xc05863ff }, .{ .start = 0xc0a80000, .end = 0xc0a8ffff }, .{ .start = 0xc6120000, .end = 0xc613ffff }, .{ .start = 0xc6336400, .end = 0xc63364ff }, @@ -243,6 +244,7 @@ test "isReserved flags RFC 6890 space, passes public IPs" { try std.testing.expect(isReserved((try parseCidr("169.254.5.5/32")).start)); try std.testing.expect(isReserved((try parseCidr("224.0.0.1/32")).start)); try std.testing.expect(isReserved((try parseCidr("0.0.0.0/32")).start)); + try std.testing.expect(isReserved((try parseCidr("192.88.99.1/32")).start)); try std.testing.expect(!isReserved((try parseCidr("8.8.8.8/32")).start)); try std.testing.expect(!isReserved((try parseCidr("1.1.1.1/32")).start)); } @@ -318,9 +320,9 @@ test "shards with a shared seed union to the full bijection with no overlap" { } test "initShard rejects nonsensical shard counts" { - const cidrs = [_]Range{try parseCidr("8.8.8.0/30")}; // 4 ips - const ports = [_]u16{80}; // total = 4, order = smallestPrimeAbove(4)-1 = 4 - try std.testing.expectError(error.InvalidShardCount, Engine.initShard(std.testing.allocator, &cidrs, &ports, 1, 0, 0)); // num_shards 0 - try std.testing.expectError(error.InvalidShardCount, Engine.initShard(std.testing.allocator, &cidrs, &ports, 1, 100, 0)); // more shards than order - try std.testing.expectError(error.InvalidShardCount, Engine.initShard(std.testing.allocator, &cidrs, &ports, 1, 2, 5)); // shard_id out of range + const cidrs = [_]Range{try parseCidr("8.8.8.0/30")}; + const ports = [_]u16{80}; + try std.testing.expectError(error.InvalidShardCount, Engine.initShard(std.testing.allocator, &cidrs, &ports, 1, 0, 0)); + try std.testing.expectError(error.InvalidShardCount, Engine.initShard(std.testing.allocator, &cidrs, &ports, 1, 100, 0)); + try std.testing.expectError(error.InvalidShardCount, Engine.initShard(std.testing.allocator, &cidrs, &ports, 1, 2, 5)); } diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/tx.zig b/PROJECTS/advanced/zig-stateless-scanner/src/tx.zig index 0fc09a41..ba0b2aa3 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/tx.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/tx.zig @@ -15,6 +15,7 @@ pub fn run( sink: anytype, clock: anytype, max_packets: u64, + deadline_ns: u64, ) u64 { _ = bucket.takeBatch(clock.now(), 0); var sent: u64 = 0; @@ -22,7 +23,9 @@ pub fn run( var pending: ?targets.Target = engine.next(); while (pending != null and sent < max_packets) { - const granted = bucket.takeBatch(clock.now(), max_packets - sent); + const now_ns = clock.now(); + if (now_ns >= deadline_ns) break; + const granted = bucket.takeBatch(now_ns, max_packets - sent); if (granted == 0) { clock.sleepNs(bucket.step_ns); continue; @@ -92,7 +95,7 @@ test "the TX engine drives the M2 bijection through stamp + ratelimit + submit" var sink = FakeSink{ .allocator = std.testing.allocator }; defer sink.frames.deinit(std.testing.allocator); - const sent = run(&eng, &tmpl, &tb, &sink, &clock, 1_000_000); + const sent = run(&eng, &tmpl, &tb, &sink, &clock, 1_000_000, std.math.maxInt(u64)); try std.testing.expectEqual(@as(u64, 8), sent); try std.testing.expectEqual(@as(usize, 8), sink.frames.items.len); try std.testing.expect(sink.kicks >= 1); @@ -130,7 +133,40 @@ test "max_packets caps the send count below the target total" { var sink = FakeSink{ .allocator = std.testing.allocator }; defer sink.frames.deinit(std.testing.allocator); - const sent = run(&eng, &tmpl, &tb, &sink, &clock, 5); + const sent = run(&eng, &tmpl, &tb, &sink, &clock, 5, std.math.maxInt(u64)); try std.testing.expectEqual(@as(u64, 5), sent); try std.testing.expectEqual(@as(usize, 5), sink.frames.items.len); } + +const StuckSink = struct { + kicks: usize = 0, + fn submit(_: *StuckSink, _: []const u8) bool { + return false; + } + fn kick(self: *StuckSink) void { + self.kicks += 1; + } +}; + +test "run bails at the deadline when the sink never drains (stall watchdog)" { + const test_key = [_]u8{0} ** 16; + const cidrs = [_]targets.Range{try targets.parseCidr("8.8.8.0/28")}; + const ports = [_]u16{80}; + var eng = try targets.Engine.init(std.testing.allocator, &cidrs, &ports, 0x1234); + defer eng.deinit(); + + const tmpl = template.SynTemplate.init(.{ + .src_mac = .{0} ** 6, + .dst_mac = .{0} ** 6, + .src_ip = 0x0a000001, + .src_port = 40000, + .cookie = cookie.Cookie.init(test_key), + }); + var tb = ratelimit.TokenBucket.init(1000, 64); + var clock = FakeClock{}; + var sink = StuckSink{}; + + const sent = run(&eng, &tmpl, &tb, &sink, &clock, 1_000_000, 5_000_000_000); + try std.testing.expectEqual(@as(u64, 0), sent); + try std.testing.expect(sink.kicks >= 1); +} diff --git a/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig b/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig index b5fe764d..e22f5f30 100644 --- a/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig +++ b/PROJECTS/advanced/zig-stateless-scanner/src/txcmd.zig @@ -22,6 +22,8 @@ const default_iface = "lo"; const default_rate: u64 = 10_000; const default_src_port: u16 = 40_000; const default_ports = [_]u16{80}; +const ns_per_sec: u64 = 1_000_000_000; +const tx_budget_floor_ns: u64 = 60 * ns_per_sec; pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) !void { var buf: [512]u8 = undefined; @@ -79,7 +81,9 @@ pub fn run(io: std.Io, allocator: std.mem.Allocator, args: []const []const u8) ! var clock = RealClock{}; const t0 = clock.now(); - const sent = tx.run(&eng, &tmpl, &bucket, &backend, &clock, count); + const est_tx_ns: u64 = if (rate > 0) (count / rate) *| ns_per_sec else tx_budget_floor_ns; + const deadline_ns = t0 +| (est_tx_ns *| 4) +| tx_budget_floor_ns; + const sent = tx.run(&eng, &tmpl, &bucket, &backend, &clock, count, deadline_ns); const elapsed_ns = clock.now() - t0; const elapsed_s = @as(f64, @floatFromInt(elapsed_ns)) / 1_000_000_000.0;