Practice (75 min)

1. Pick one real finding from an earlier Part's hands-on lab (cloud privesc, k8s escalation, a web-app chain). Write its full Alerting and Detection Strategy: hypothesis, ATT&CK mapping, data source, expected false positives, priority, response.
2. Write it as a Sigma rule. Build two log fixtures (one that should match, one realistic-but-benign that shouldn't) and a unit test that asserts both.
3. Compile the Sigma rule to your actual SIEM/log-query backend with sigma-cli; deploy it.
4. Live-validate: re-run the original attack technique (Atomic Red Team / Stratus Red Team / your own lab repro) and confirm the deployed rule fires against real telemetry, recording latency.
5. Deliberately introduce a schema change in your test log source (rename a field) and confirm the rule silently stops matching — this is the maintenance failure mode to design monitoring against (e.g. a "rule hasn't fired AND hasn't been re-validated in N days" meta-alert).
6. Start (or add to) your ATT&CK coverage heat map (Part 1 Ch 2) with this detection's technique marked covered, with a link to its Sigma rule and its last live-validation date.

Build it in Go (40 min) — detections as code, with tests that gate deployment

The lifecycle above says detections are code: versioned, reviewed, tested, deployed. This lab makes that literal. Each rule is a Go type (including a stateful one), each ships with fixtures it must and must not fire on, and the program refuses to run if any rule fails its own tests.

// detect: detection-as-code (Chapter 51). Rules are Go functions over a stream
// of JSON events; every rule ships with fixtures it MUST fire on and MUST NOT
// fire on, and the program refuses to run if any rule fails its own tests --
// exactly the gate a CI pipeline for detections enforces.
//
//	go run ./detect < events.jsonl     (or with no input: runs the built-in demo stream)
package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"time"
)

type event struct {
	Time    time.Time         `json:"time"`
	Action  string            `json:"action"` // login, kms:Decrypt, iam:CreateAccessKey, ...
	Actor   string            `json:"actor"`
	SrcIP   string            `json:"src_ip"`
	Outcome string            `json:"outcome"` // success | failure | denied
	Context map[string]string `json:"context"`
}

type alert struct{ rule, severity, detail string }

// A rule sees events in time order and may keep state between them.
type rule interface {
	name() string
	observe(e event) *alert
}

// ---------------------------------------------------------------- rules --

// bruteThenSuccess: >= 5 failed logins for one actor within 2 minutes,
// followed by a success -- a guessed password, not just noise.
type bruteThenSuccess struct{ fails map[string][]time.Time }

func (r *bruteThenSuccess) name() string { return "brute-force-then-success" }
func (r *bruteThenSuccess) observe(e event) *alert {
	if e.Action != "login" {
		return nil
	}
	if r.fails == nil {
		r.fails = map[string][]time.Time{}
	}
	if e.Outcome == "failure" {
		r.fails[e.Actor] = append(r.fails[e.Actor], e.Time)
		return nil
	}
	recent := 0
	for _, t := range r.fails[e.Actor] {
		if e.Time.Sub(t) <= 2*time.Minute {
			recent++
		}
	}
	delete(r.fails, e.Actor)
	if recent >= 5 {
		return &alert{r.name(), "high", fmt.Sprintf("%s logged in from %s after %d failures in 2m", e.Actor, e.SrcIP, recent)}
	}
	return nil
}

// crossTenantKey: a KMS decrypt DENIED because the caller tried another
// tenant's key (Chapter 9's worked example). Almost never benign.
type crossTenantKey struct{}

func (crossTenantKey) name() string { return "cross-tenant-key-access" }
func (crossTenantKey) observe(e event) *alert {
	if e.Action == "kms:Decrypt" && e.Outcome == "denied" && e.Context["tenant"] != "" {
		return &alert{"cross-tenant-key-access", "high",
			fmt.Sprintf("%s was denied tenant %s's key", e.Actor, e.Context["tenant"])}
	}
	return nil
}

// newKeyForHuman: long-lived access keys created for a human user (policy:
// humans use SSO). Low severity on its own; high in an attack-path chain.
type newKeyForHuman struct{}

func (newKeyForHuman) name() string { return "access-key-for-human" }
func (newKeyForHuman) observe(e event) *alert {
	if e.Action == "iam:CreateAccessKey" && e.Outcome == "success" && strings.HasPrefix(e.Context["user"], "human/") {
		return &alert{"access-key-for-human", "medium", fmt.Sprintf("%s created a key for %s", e.Actor, e.Context["user"])}
	}
	return nil
}

// ------------------------------------------------- fixtures: rule tests --

type fixture struct {
	rule   func() rule
	events []event
	want   bool // must the LAST event fire?
}

func at(min int) time.Time { return time.Date(2026, 10, 3, 9, min, 0, 0, time.UTC) }

func login(min int, actor, outcome string) event {
	return event{Time: at(min), Action: "login", Actor: actor, SrcIP: "203.0.113.7", Outcome: outcome}
}

var fixtures = []fixture{
	{func() rule { return &bruteThenSuccess{} }, append(repeat(login(0, "alice", "failure"), 5), login(1, "alice", "success")), true},
	{func() rule { return &bruteThenSuccess{} }, append(repeat(login(0, "alice", "failure"), 2), login(1, "alice", "success")), false},
	{func() rule { return &bruteThenSuccess{} }, append(repeat(login(0, "alice", "failure"), 5), login(9, "alice", "success")), false}, // too slow
	{func() rule { return crossTenantKey{} }, []event{{Action: "kms:Decrypt", Outcome: "denied", Context: map[string]string{"tenant": "99"}}}, true},
	{func() rule { return crossTenantKey{} }, []event{{Action: "kms:Decrypt", Outcome: "success", Context: map[string]string{"tenant": "99"}}}, false},
	{func() rule { return newKeyForHuman{} }, []event{{Action: "iam:CreateAccessKey", Outcome: "success", Context: map[string]string{"user": "human/bob"}}}, true},
	{func() rule { return newKeyForHuman{} }, []event{{Action: "iam:CreateAccessKey", Outcome: "success", Context: map[string]string{"user": "svc/ci"}}}, false},
}

func repeat(e event, n int) []event {
	out := make([]event, n)
	for i := range out {
		out[i] = e
	}
	return out
}

func testRules() bool {
	ok := true
	for i, f := range fixtures {
		r := f.rule()
		var last *alert
		for _, e := range f.events {
			last = r.observe(e)
		}
		if (last != nil) != f.want {
			fmt.Printf("FAIL fixture %d (%s): want fire=%v\n", i, r.name(), f.want)
			ok = false
		}
	}
	fmt.Printf("rule tests: %d fixtures, all passed=%v\n\n", len(fixtures), ok)
	return ok
}

// ---------------------------------------------------------------- main --

func main() {
	if !testRules() {
		os.Exit(1) // never deploy a detection that fails its own fixtures
	}
	rules := []rule{&bruteThenSuccess{}, crossTenantKey{}, newKeyForHuman{}}

	var events []event
	if st, _ := os.Stdin.Stat(); st.Mode()&os.ModeCharDevice == 0 {
		sc := bufio.NewScanner(os.Stdin)
		for sc.Scan() {
			var e event
			if json.Unmarshal(sc.Bytes(), &e) == nil {
				events = append(events, e)
			}
		}
	} else { // a small attack story: guessing, a foothold, persistence, then data
		events = append(repeat(login(0, "carol", "failure"), 6), login(1, "carol", "success"),
			event{Time: at(3), Action: "iam:CreateAccessKey", Actor: "carol", Outcome: "success", Context: map[string]string{"user": "human/carol"}},
			event{Time: at(5), Action: "kms:Decrypt", Actor: "carol", Outcome: "denied", Context: map[string]string{"tenant": "42"}},
			event{Time: at(6), Action: "login", Actor: "dave", Outcome: "success"})
	}
	for _, e := range events {
		for _, r := range rules {
			if a := r.observe(e); a != nil {
				fmt.Printf("%s [%-6s] %-26s %s\n", e.Time.Format("15:04"), a.severity, a.rule, a.detail)
			}
		}
	}
}

$ go run ./detect
rule tests: 7 fixtures, all passed=true

09:01 [high  ] brute-force-then-success   carol logged in from 203.0.113.7 after 6 failures in 2m
09:03 [medium] access-key-for-human       carol created a key for human/carol
09:05 [high  ] cross-tenant-key-access    carol was denied tenant 42's key

$ echo '{"time":"2026-10-03T10:00:00Z","action":"kms:Decrypt","actor":"analytics-role","outcome":"denied","context":{"tenant":"99"}}' | go run ./detect
rule tests: 7 fixtures, all passed=true

10:00 [high  ] cross-tenant-key-access    analytics-role was denied tenant 99's key

What to notice:

• The fixtures encode the edge cases, such as "5 failures, then success 8 minutes later" must not fire. Analysts argue about thresholds in fixtures, in code review, not in production pages.
• The demo stream reads as a story: credential guessing → persistence (a long-lived key) → probing data (a denied cross-tenant key). Three medium-to-high alerts in five minutes on one actor is the correlation Chapter 54 asks for.
• The second command is the Chapter 9 worked example, firing on real JSON. Ship your CloudTrail, audit-log, or SIEM export to stdin in the same shape.

Exercises:

1. Add an impossible-travel rule (two logins for one user, with geo-IP distance ÷ time > 1,000 km/h) together with its fixtures.
2. Add a correlation rule that raises one critical alert when one actor triggers three different rules within 15 minutes.
3. Run the fixtures in CI as go test and block merges that break them.