Security, Data Protection, Authorization, and Safe Load-Test Boundaries: Guided Hands-On Workflow

1. Lab layout and assumptions

p31-lab/
├── scope/authorization.json
├── fixtures/safe_target.py
├── plans/safe-load.jmx
├── config/safe.properties
├── tools/safe_launcher.py
├── tools/scan_artifacts.py
├── tools/cleanup_audit.py
└── results/

2. Write authorization before code

scope/authorization.json:

{
  "schema_version": 1,
  "authorization_id": "P31-LOCAL-LAB-ONLY",
  "environment": "local-disposable",
  "owner": "learner-controlled-local-fixture",
  "purpose": "Apache JMeter security-boundary training",
  "authorization_window": "only while the local fixture process is running",
  "target": {
    "base_url": "http://127.0.0.1:8031",
    "scheme": "http",
    "host": "127.0.0.1",
    "port": 8031,
    "allowed_paths": ["/work"],
    "allowed_methods": ["POST"]
  },
  "workload": {
    "max_threads": 3,
    "max_loops": 10,
    "max_samples": 30,
    "min_pacing_ms": 150,
    "max_duration_s": 15,
    "target_max_rps": 25
  },
  "data": {
    "classification": "synthetic-only",
    "production_customer_data": false
  },
  "secret": {
    "source": "environment:P31_FAKE_TOKEN",
    "classification": "fake-training-secret",
    "persist_in_jmx": false,
    "persist_in_results": false
  },
  "evidence": {
    "retention_days": 7,
    "forbidden_saved_fields": [
      "requestHeaders",
      "responseHeaders",
      "responseData",
      "samplerData",
      "URL"
    ]
  }
}

This local manifest is deliberately narrow. In a real organization add the actual approver/change record, time window, environment owner, dependency exclusions and emergency contact. “Local fixture process is running” is the lab's authorization window because the learner owns both ends.

3. Build the safe target

fixtures/safe_target.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse
from collections import deque
import argparse, hashlib, hmac, json, os, re, threading, time

FIXTURE_VERSION = "prompt31-safe-target-v1"
SAFE_TEXT = re.compile(r"^[A-Za-z0-9_.:@-]{1,96}$")

lock = threading.Lock()
event_log = None
expected_token = None
max_requests = 30
max_rps = 25
recent = deque()
state = {
    "work_attempts": 0,
    "successes": 0,
    "auth_failures": 0,
    "scope_rejections": 0,
    "rate_rejections": 0,
}

def token_fp(value):
    return hashlib.sha256(value.encode()).hexdigest()[:12]

def snapshot():
    with lock:
        return {
            "fixture_version": FIXTURE_VERSION,
            "max_requests": max_requests,
            "max_rps": max_rps,
            "expected_token_fp": token_fp(expected_token),
            **state,
        }

def write_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as h:
            h.write(json.dumps(event, sort_keys=True) + "\n")

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def send_json(self, status, payload):
        raw = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()
        self.wfile.write(raw)

    def do_GET(self):
        if urlparse(self.path).path == "/health":
            self.send_json(200, {"status": "ok", "state": snapshot()})
            return
        if urlparse(self.path).path == "/stats":
            self.send_json(200, {"status": "ok", "state": snapshot()})
            return
        self.send_json(404, {"status": "not_found"})

    def do_POST(self):
        started = int(time.time() * 1000)
        path = urlparse(self.path).path
        if path != "/work":
            with lock:
                state["scope_rejections"] += 1
            self.send_json(404, {"status": "not_allowed"})
            return

        with lock:
            state["work_attempts"] += 1
            attempt = state["work_attempts"]
            now = time.monotonic()
            while recent and now - recent[0] >= 1.0:
                recent.popleft()
            over_rate = len(recent) >= max_rps
            if not over_rate:
                recent.append(now)

        if attempt > max_requests:
            with lock:
                state["scope_rejections"] += 1
            self.send_json(429, {"status": "sample_ceiling"})
            write_event({
                "ts_ms": int(time.time() * 1000),
                "operation": "work", "status": 429,
                "reason": "sample_ceiling",
                "authorization": "Bearer <redacted>" if self.headers.get("Authorization") else None
            })
            return

        if over_rate:
            with lock:
                state["rate_rejections"] += 1
            self.send_json(429, {"status": "rate_ceiling"})
            write_event({
                "ts_ms": int(time.time() * 1000),
                "operation": "work", "status": 429,
                "reason": "rate_ceiling",
                "authorization": "Bearer <redacted>" if self.headers.get("Authorization") else None
            })
            return

        auth = self.headers.get("Authorization", "")
        supplied = auth[7:] if auth.startswith("Bearer ") else ""
        if not supplied or not hmac.compare_digest(supplied, expected_token):
            with lock:
                state["auth_failures"] += 1
            self.send_json(401, {"status": "unauthorized"})
            write_event({
                "ts_ms": int(time.time() * 1000),
                "operation": "work", "status": 401,
                "authorization": "Bearer <redacted>" if auth else None,
                "token_fp": token_fp(supplied) if supplied else None,
            })
            return

        scope_id = self.headers.get("X-Load-Test-Scope", "")
        if scope_id != "P31-LOCAL-LAB-ONLY":
            with lock:
                state["scope_rejections"] += 1
            self.send_json(403, {"status": "scope_header_required"})
            return

        length = int(self.headers.get("Content-Length", "0") or 0)
        if length <= 0 or length > 1024:
            self.send_json(400, {"status": "invalid_body_size"})
            return

        raw = self.rfile.read(length)
        try:
            body = json.loads(raw)
        except Exception:
            self.send_json(400, {"status": "invalid_json"})
            return

        allowed_keys = {"synthetic_user", "request_id", "note"}
        if set(body) != allowed_keys:
            self.send_json(400, {"status": "invalid_fields"})
            return
        if not SAFE_TEXT.fullmatch(str(body["synthetic_user"])):
            self.send_json(400, {"status": "invalid_synthetic_user"})
            return
        if not SAFE_TEXT.fullmatch(str(body["request_id"])):
            self.send_json(400, {"status": "invalid_request_id"})
            return

        time.sleep(0.025)
        with lock:
            state["successes"] += 1
        ended = int(time.time() * 1000)

        self.send_json(200, {
            "status": "ok",
            "request_id": body["request_id"],
            "synthetic": True,
        })

        # Deliberately retain only redacted/derived evidence:
        # no raw Authorization token and no raw request body/body values.
        write_event({
            "ts_ms": ended,
            "operation": "work",
            "status": 200,
            "authorization": "Bearer <redacted>",
            "token_fp": token_fp(supplied),
            "body_sha256": hashlib.sha256(raw).hexdigest(),
            "body_fields": sorted(body.keys()),
            "body_bytes": len(raw),
            "service_wall_ms": ended - started,
        })

    def log_message(self, format, *args):
        return

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=8031)
    p.add_argument("--max-requests", type=int, default=30)
    p.add_argument("--max-rps", type=int, default=25)
    p.add_argument("--log", required=True)
    args = p.parse_args()

    if args.host != "127.0.0.1":
        raise SystemExit("lab fixture may bind only to 127.0.0.1")
    if not 1 <= args.max_requests <= 30:
        raise SystemExit("max-requests must be 1..30")
    if not 1 <= args.max_rps <= 25:
        raise SystemExit("max-rps must be 1..25")

    token = os.environ.get("P31_FAKE_TOKEN", "")
    if not token.startswith("p31-fake-") or len(token) < 16:
        raise SystemExit("P31_FAKE_TOKEN must contain the fake training token")

    global event_log, expected_token, max_requests, max_rps
    expected_token = token
    max_requests = args.max_requests
    max_rps = args.max_rps

    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    print(f"fixture_version={FIXTURE_VERSION}", flush=True)
    print(f"listen=http://127.0.0.1:{args.port}", flush=True)
    print(f"expected_token_fp={token_fp(expected_token)}", flush=True)
    print(f"max_requests={max_requests} max_rps={max_rps}", flush=True)
    ThreadingHTTPServer((args.host, args.port), Handler).serve_forever()

if __name__ == "__main__":
    main()

The fixture reads only the fake token from its process environment, binds only to loopback, rejects more than30 attempts and more than25 requests/s, requires the scope header, and logs only a token fingerprint/body hash/body field names—not the raw Authorization value or request body.

4. Source the fake secret without printing it

PowerShell:

$env:P31_FAKE_TOKEN = "p31-fake-token-4e65d5cf"
"Fake training token is set; value intentionally not printed again."

Bash:

export P31_FAKE_TOKEN='p31-fake-token-4e65d5cf'
echo 'Fake training token is set; value intentionally not printed again.'

This literal is intentionally fake and disposable. Do not copy a real secret into the tutorial.

5. Start and inspect the target

python .\fixtures\safe_target.py `
  --host 127.0.0.1 --port 8031 `
  --max-requests 30 --max-rps 25 `
  --log .\results\target-events.jsonl

In another shell:

curl.exe --fail --silent http://127.0.0.1:8031/health

Require loopback/30/25 and the expected token fingerprint. Health does not expose the raw token.

6. Pin lean/redacted result policy

config/safe.properties:

target.host=127.0.0.1
target.port=8031
threads=3
loops=10
pacing.ms=200
connect.timeout.ms=500
response.timeout.ms=1500

jmeter.httpsampler=HttpClient4
httpclient4.retrycount=0

jmeter.save.saveservice.output_format=csv
jmeter.save.saveservice.print_field_names=true
jmeter.save.saveservice.timestamp_format=ms
jmeter.save.saveservice.time=true
jmeter.save.saveservice.label=true
jmeter.save.saveservice.response_code=true
jmeter.save.saveservice.response_message=true
jmeter.save.saveservice.thread_name=true
jmeter.save.saveservice.successful=true
jmeter.save.saveservice.bytes=true
jmeter.save.saveservice.sent_bytes=true
jmeter.save.saveservice.thread_counts=true
jmeter.save.saveservice.latency=true
jmeter.save.saveservice.connect_time=true
jmeter.save.saveservice.assertion_results_failure_message=true

# Deliberate data-minimization boundary:
jmeter.save.saveservice.response_data=false
jmeter.save.saveservice.response_data.on_error=false
jmeter.save.saveservice.samplerData=false
jmeter.save.saveservice.responseHeaders=false
jmeter.save.saveservice.requestHeaders=false
jmeter.save.saveservice.url=false
sample_variables=

Even though current JMeter defaults already avoid several sensitive fields, the lab pins them explicitly. Retained CSV keeps performance/result evidence while excluding request/response headers, bodies, sampler data and URL.

7. Author the bounded JMX

JMX tree:

Test Plan
├── HTTP Request Defaults
│   host=${__P(target.host,127.0.0.1)}
│   port=${__P(target.port,8031)}
│   implementation=HttpClient4
│   connect timeout=${__P(connect.timeout.ms,500)}
│   response timeout=${__P(response.timeout.ms,1500)}
├── HTTP Header Manager
│   Content-Type: application/json
│   Authorization: Bearer ${__P(fake.token)}
│   X-Load-Test-Scope: P31-LOCAL-LAB-ONLY
└── Thread Group — Safe bounded lab
    threads=${__P(threads,3)}
    loops=${__P(loops,10)}
    Action after Sampler error=Stop Thread
    ├── Counter -> SEQ (per user)
    └── HTTP Request — SafeWork
        POST /work
        Body Data:
          {
            "synthetic_user": "synthetic-user-${__threadNum}",
            "request_id": "${__threadNum}-${SEQ}",
            "note": "non-sensitive-training-data"
          }
        Use KeepAlive=checked
        ├── Constant Timer ${__P(pacing.ms,200)} ms
        └── Response Assertion: HTTP response code = 200

The JMX reads only properties. It does not store the fake token. Stop Thread on sampler error prevents a bad auth/rate condition from continuing blindly; the launcher/target ceilings remain the stronger external boundaries.

8. Build the pre-network launcher

tools/safe_launcher.py:

import argparse, hashlib, json, os, shutil, stat, subprocess, tempfile, time, urllib.request
from pathlib import Path
from urllib.parse import urlsplit

EXIT_SCOPE_DENIED = 30
EXIT_SECRET_MISSING = 31
EXIT_WORKLOAD_DENIED = 32
EXIT_TARGET_PREFLIGHT = 33
EXIT_JMETER_FAIL = 34
EXIT_TIMEOUT = 35

def sha256(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def write_guard(out, doc):
    out.mkdir(parents=True, exist_ok=True)
    (out / "guard.json").write_text(json.dumps(doc, indent=2), encoding="utf-8")

def resolve_jmeter(explicit=None):
    if explicit:
        return Path(explicit).resolve()
    if os.environ.get("JMETER_BIN"):
        return Path(os.environ["JMETER_BIN"]).resolve()
    if os.environ.get("JMETER_HOME"):
        name = "jmeter.bat" if os.name == "nt" else "jmeter"
        return (Path(os.environ["JMETER_HOME"]) / "bin" / name).resolve()
    found = shutil.which("jmeter.bat" if os.name == "nt" else "jmeter")
    if found:
        return Path(found).resolve()
    raise RuntimeError("JMeter not found")

def run_cmd(exe, args, **kwargs):
    exe = str(exe)
    if os.name == "nt" and exe.lower().endswith((".bat", ".cmd")):
        cmdline = subprocess.list2cmdline([exe] + [str(x) for x in args])
        return subprocess.run(["cmd.exe", "/d", "/s", "/c", cmdline], **kwargs)
    return subprocess.run([exe] + [str(x) for x in args], **kwargs)

def validate_target(scope, target_url):
    expected = scope["target"]["base_url"].rstrip("/")
    parsed = urlsplit(target_url)
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        return False, "credentials/query/fragment are not allowed in target URL"
    if target_url.rstrip("/") != expected:
        return False, f"target {target_url!r} is not exact allow-listed target {expected!r}"
    if parsed.scheme != scope["target"]["scheme"]:
        return False, "scheme mismatch"
    if parsed.hostname != scope["target"]["host"]:
        return False, "host mismatch"
    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    if port != int(scope["target"]["port"]):
        return False, "port mismatch"
    return True, "allowed"

def validate_workload(scope, threads, loops, pacing_ms):
    w = scope["workload"]
    samples = threads * loops
    failures = []
    if threads < 1 or threads > int(w["max_threads"]):
        failures.append("threads outside authorization")
    if loops < 1 or loops > int(w["max_loops"]):
        failures.append("loops outside authorization")
    if samples > int(w["max_samples"]):
        failures.append("sample ceiling exceeded")
    if pacing_ms < int(w["min_pacing_ms"]):
        failures.append("pacing below authorized minimum")
    return failures, samples

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--scope", required=True)
    p.add_argument("--target-url", required=True)
    p.add_argument("--jmx", required=True)
    p.add_argument("--properties", required=True)
    p.add_argument("--threads", type=int, default=3)
    p.add_argument("--loops", type=int, default=10)
    p.add_argument("--pacing-ms", type=int, default=200)
    p.add_argument("--out", required=True)
    p.add_argument("--jmeter-bin")
    args = p.parse_args()

    out = Path(args.out).resolve()
    scope_path = Path(args.scope).resolve()
    jmx = Path(args.jmx).resolve()
    props = Path(args.properties).resolve()
    scope = json.loads(scope_path.read_text(encoding="utf-8"))

    doc = {
        "schema_version": 1,
        "authorization_id": scope.get("authorization_id"),
        "target_url": args.target_url,
        "network_attempted": False,
        "jmeter_invoked": False,
        "temporary_secret_file_deleted": None,
        "status": None,
        "reason": None,
    }

    allowed, reason = validate_target(scope, args.target_url)
    if not allowed:
        doc.update(status="SCOPE_DENIED", reason=reason)
        write_guard(out, doc)
        return EXIT_SCOPE_DENIED

    workload_failures, samples = validate_workload(
        scope, args.threads, args.loops, args.pacing_ms
    )
    doc["configured_samples"] = samples
    if workload_failures:
        doc.update(status="WORKLOAD_DENIED", reason="; ".join(workload_failures))
        write_guard(out, doc)
        return EXIT_WORKLOAD_DENIED

    token = os.environ.get("P31_FAKE_TOKEN", "")
    if not token.startswith("p31-fake-") or len(token) < 16:
        doc.update(status="SECRET_MISSING", reason="fake training token missing")
        write_guard(out, doc)
        return EXIT_SECRET_MISSING

    # Only now, after scope + workload + secret validation, may any network call happen.
    try:
        doc["network_attempted"] = True
        with urllib.request.urlopen(args.target_url.rstrip("/") + "/health", timeout=2) as r:
            health = json.loads(r.read())
        if health.get("status") != "ok":
            raise RuntimeError("target health is not ok")
        (out / "preflight.json").parent.mkdir(parents=True, exist_ok=True)
        (out / "preflight.json").write_text(json.dumps(health, indent=2), encoding="utf-8")
    except Exception as exc:
        doc.update(status="TARGET_PREFLIGHT_FAILED", reason=str(exc))
        write_guard(out, doc)
        return EXIT_TARGET_PREFLIGHT

    try:
        jmeter = resolve_jmeter(args.jmeter_bin)
    except Exception as exc:
        doc.update(status="JMETER_NOT_FOUND", reason=str(exc))
        write_guard(out, doc)
        return EXIT_JMETER_FAIL

    out.mkdir(parents=True, exist_ok=True)
    manifest = {
        "scope_sha256": sha256(scope_path),
        "jmx_sha256": sha256(jmx),
        "properties_sha256": sha256(props),
        "configured_threads": args.threads,
        "configured_loops": args.loops,
        "configured_samples": samples,
        "pacing_ms": args.pacing_ms,
        "target_url": args.target_url,
    }
    (out / "input-manifest.json").write_text(
        json.dumps(manifest, indent=2), encoding="utf-8"
    )

    dashboard = out / "dashboard"
    if dashboard.exists():
        shutil.rmtree(dashboard)

    temp_path = None
    started = time.monotonic()
    try:
        with tempfile.TemporaryDirectory(prefix="p31-secret-") as td:
            temp_path = Path(td) / "runtime-secret.properties"
            parsed = urlsplit(args.target_url)
            runtime_props = (
                f"fake.token={token}\n"
                f"target.host={parsed.hostname}\n"
                f"target.port={parsed.port}\n"
                f"threads={args.threads}\n"
                f"loops={args.loops}\n"
                f"pacing.ms={args.pacing_ms}\n"
            )
            temp_path.write_text(runtime_props, encoding="utf-8")
            if os.name != "nt":
                os.chmod(temp_path, stat.S_IRUSR | stat.S_IWUSR)

            child_env = os.environ.copy()
            child_env.pop("P31_FAKE_TOKEN", None)

            cmd_args = [
                "-n", "-t", str(jmx),
                "-q", str(props),
                "-q", str(temp_path),
                "-l", str(out / "results.jtl"),
                "-j", str(out / "jmeter.log"),
                "-e", "-o", str(dashboard),
            ]
            doc["jmeter_invoked"] = True
            with (out / "jmeter-console.log").open("w", encoding="utf-8") as console:
                try:
                    cp = run_cmd(
                        jmeter, cmd_args,
                        stdout=console, stderr=subprocess.STDOUT,
                        text=True,
                        timeout=int(scope["workload"]["max_duration_s"]),
                        env=child_env,
                    )
                except subprocess.TimeoutExpired:
                    doc.update(status="TIMEOUT_ABORT", reason="JMeter exceeded authorized duration")
                    return EXIT_TIMEOUT

            if cp.returncode != 0:
                doc.update(status="JMETER_FAILED", reason=f"exit={cp.returncode}")
                return EXIT_JMETER_FAIL

            doc.update(status="PASS", reason="authorized bounded run completed")
            return 0
    finally:
        doc["duration_s"] = round(time.monotonic() - started, 3)
        doc["temporary_secret_file_deleted"] = (
            True if temp_path is None else not temp_path.exists()
        )
        write_guard(out, doc)

if __name__ == "__main__":
    raise SystemExit(main())

Order is the core security property: exact target → workload bounds → fake secret presence → network health preflight → JMeter resolution/invocation. A denied target writes guard.json with network_attempted=false/jmeter_invoked=false.

9. Prove an unauthorized target fails before traffic

The URL below is only an input string; the guard must reject it before DNS/network/JMeter:

python .\tools\safe_launcher.py `
  --scope .\scope\authorization.json `
  --target-url https://example.com `
  --jmx .\plans\safe-load.jmx `
  --properties .\config\safe.properties `
  --threads 3 --loops 10 --pacing-ms 200 `
  --out .\results\rejected-target

if ($LASTEXITCODE -ne 30) {
  throw "Expected SCOPE_DENIED exit 30"
}

Get-Content .\results\rejected-target\guard.json

Require: status=SCOPE_DENIED, network_attempted=false, jmeter_invoked=false. Check target /stats: work_attempts must still be zero.

10. Run the authorized bounded workload

python .\tools\safe_launcher.py `
  --scope .\scope\authorization.json `
  --target-url http://127.0.0.1:8031 `
  --jmx .\plans\safe-load.jmx `
  --properties .\config\safe.properties `
  --threads 3 --loops 10 --pacing-ms 200 `
  --out .\results\authorized

if ($LASTEXITCODE -ne 0) {
  throw "Authorized run failed"
}

The launcher creates a private temporary properties file containing the fake token, removes the token from JMeter's child environment, runs CLI mode, then deletes the temporary directory. No token appears in CLI arguments.

11. Verify achieved workload and target kill boundaries

curl.exe --fail --silent http://127.0.0.1:8031/stats
Get-Content .\results\authorized\guard.json

Expected: 30 work attempts/successes, zero auth/scope/rate failures, launcher PASS and temporary_secret_file_deleted=true. If any count exceeds30, stop and preserve evidence.

12. Scan retained evidence for leaks

tools/scan_artifacts.py:

import argparse, json, os, re
from pathlib import Path

TEXT_EXTS = {".jtl", ".log", ".json", ".jsonl", ".txt", ".csv", ".html", ".properties"}
RAW_AUTH = re.compile(r"Authorization\s*[:=]\s*Bearer\s+(?!<redacted>)[^\s\"',}]+", re.I)

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--root", required=True)
    p.add_argument("--secret-env", default="P31_FAKE_TOKEN")
    p.add_argument("--out", required=True)
    args = p.parse_args()

    root = Path(args.root).resolve()
    secret = os.environ.get(args.secret_env, "")
    findings = []

    for path in root.rglob("*"):
        if not path.is_file() or path.suffix.lower() not in TEXT_EXTS:
            continue
        text = path.read_text(encoding="utf-8", errors="replace")
        rel = str(path.relative_to(root))
        if secret and secret in text:
            findings.append({"file": rel, "type": "raw_fake_secret"})
        if RAW_AUTH.search(text):
            findings.append({"file": rel, "type": "raw_authorization_header"})
        if path.name.endswith(".jsonl"):
            for line_no, line in enumerate(text.splitlines(), 1):
                if not line.strip():
                    continue
                try:
                    doc = json.loads(line)
                except Exception:
                    continue
                forbidden = {"body", "raw_body", "authorization_raw", "request_headers"}
                if forbidden.intersection(doc):
                    findings.append({
                        "file": rel, "line": line_no,
                        "type": "forbidden_target_event_field",
                        "fields": sorted(forbidden.intersection(doc))
                    })

    jtls = list(root.rglob("*.jtl"))
    for jtl in jtls:
        header = jtl.read_text(encoding="utf-8", errors="replace").splitlines()
        if header:
            forbidden_cols = {"requestHeaders", "responseHeaders", "responseData", "samplerData", "URL"}
            cols = set(header[0].split(","))
            bad = sorted(forbidden_cols.intersection(cols))
            if bad:
                findings.append({
                    "file": str(jtl.relative_to(root)),
                    "type": "forbidden_jtl_columns",
                    "columns": bad
                })

    result = {
        "root": str(root),
        "files_scanned": sum(1 for p in root.rglob("*") if p.is_file()),
        "findings": findings,
        "status": "PASS" if not findings else "FAIL"
    }
    Path(args.out).write_text(json.dumps(result, indent=2), encoding="utf-8")
    print(json.dumps(result, indent=2))
    raise SystemExit(0 if not findings else 4)

if __name__ == "__main__":
    main()

python .\tools\scan_artifacts.py `
  --root .\results\authorized `
  --secret-env P31_FAKE_TOKEN `
  --out .\results\authorized\redaction-scan.json

Require PASS: no raw fake token, raw bearer header, forbidden target-event body/header fields, or forbidden JTL columns.

13. Cleanup/audit record

After stopping the fixture, save tools/cleanup_audit.py:

import argparse, json, socket
from pathlib import Path

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=8031)
    p.add_argument("--guard", required=True)
    p.add_argument("--redaction-scan", required=True)
    p.add_argument("--out", required=True)
    args = p.parse_args()

    guard = json.loads(Path(args.guard).read_text(encoding="utf-8"))
    scan = json.loads(Path(args.redaction_scan).read_text(encoding="utf-8"))

    s = socket.socket()
    s.settimeout(0.5)
    try:
        port_open = s.connect_ex((args.host, args.port)) == 0
    finally:
        s.close()

    result = {
        "target_port_closed_after_cleanup": not port_open,
        "temporary_secret_file_deleted": guard.get("temporary_secret_file_deleted"),
        "redaction_scan_status": scan.get("status"),
        "authorized_run_status": guard.get("status"),
    }
    result["status"] = (
        "PASS"
        if result["target_port_closed_after_cleanup"]
        and result["temporary_secret_file_deleted"] is True
        and result["redaction_scan_status"] == "PASS"
        and result["authorized_run_status"] == "PASS"
        else "FAIL"
    )
    Path(args.out).write_text(json.dumps(result, indent=2), encoding="utf-8")
    print(json.dumps(result, indent=2))
    raise SystemExit(0 if result["status"] == "PASS" else 5)

if __name__ == "__main__":
    main()

# Stop the fixture process first, then:
python .\tools\cleanup_audit.py `
  --host 127.0.0.1 --port 8031 `
  --guard .\results\authorized\guard.json `
  --redaction-scan .\results\authorized\redaction-scan.json `
  --out .\results\cleanup-audit.json

Remove-Item Env:P31_FAKE_TOKEN

Require target port closed, temporary secret file deleted, redaction PASS and authorized launcher PASS.

14. Challenge

A team asks to point the same plan at staging.example.internal and says “it's staging, so it's automatically safe.” What must change first?

The authorization/scope record—not the JMX. The new target needs explicit owner approval, exact hostname/port/path, dependency/data classification, workload ceiling, monitoring/abort contacts, TLS trust strategy and artifact policy before it can enter the allow-list.

Knowledge check

Official references and version notes

• Apache JMeter downloads — current stable JMeter 5.6.3 and Java 8+ requirement.
• JMeter current changes — Java 17+ recommendation for the 5.6.x line.
• Apache JMeter Security Model — JMX is trusted input and may execute arbitrary code; isolate untrusted plans.
• JMeter Component Reference — Authorization Manager passwords are stored unencrypted in the test plan; recorder certificate behavior and SSL components.
• JMeter Properties Reference — result-save fields and RMI SSL settings.
• JMeter Remote Testing — RMI SSL defaults and keystore setup.
• HTTP(S) Test Script Recorder — recording workflow and temporary CA trust requirements.