Classroom Glossary Public page

Lab 7: Tool v0.3, Logging + Idempotency

649 words

~3 hr. Extend Tool v0.2 to Tool v0.3: structured JSON run logs to disk; human-readable stderr summary; idempotency check via extended fingerprint; documented rollback path. v0.3 is what becomes v1.0 in the capstone.

Goal: ship sb6141_csrf/v03.py with three new properties beyond v0.2: per-run JSON log file at ~/.sb6141-csrf/runs/<ISO8601>.json; human-readable stderr summary; idempotency check that detects already-reset state; rollback documentation (and implementation stub for the SB6141 case).

Estimated time: ~3 hr.

Prerequisites: Week 7 lecture. Lab 5 complete (v0.2 working).

Authorization line: Required in lab notebook before any --no-dry-run invocation:

Lab 7 session, 2026-MM-DD HH:MM. Target SB6141 unit <serial>; scope per signed cohort
authorization; Tool v0.3 development + testing (expect ~3 factory-resets across the
session to verify idempotency check and rollback path).

Setup

mkdir -p ~/adv-101/lab-7/sb6141_csrf ~/adv-101/lab-7/tests
cd ~/adv-101/lab-7
cp ../lab-5/sb6141_csrf/v02.py sb6141_csrf/v03.py
cp ../lab-5/sb6141_csrf/__init__.py sb6141_csrf/__init__.py
cp ../lab-5/tests/test_v02.py tests/test_v03.py

Create the runs directory:

mkdir -p ~/.sb6141-csrf/runs

Part A: Add the structured JSON log writer (~45 min)

In sb6141_csrf/v03.py, add a function that writes a per-run JSON log file:

import json
import os
from datetime import datetime, timezone
from pathlib import Path

TOOL_VERSION = '0.3'
RUNS_DIR = Path.home() / '.sb6141-csrf' / 'runs'

def write_run_log(payload: dict) -> Path:
    """Write a per-run JSON log file. Returns the file path."""
    RUNS_DIR.mkdir(parents=True, exist_ok=True)
    iso = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H-%M-%SZ')
    path = RUNS_DIR / f'{iso}.json'
    with path.open('w', encoding='utf-8') as f:
        json.dump(payload, f, indent=2, sort_keys=True)
    return path

def build_run_payload(args, fingerprint_result: dict, action_result: dict | None, exit_code: int) -> dict:
    """Construct the per-run JSON payload."""
    return {
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'tool_version': TOOL_VERSION,
        'target': args.target,
        'action': ACTION_NAME,
        'authorized_by': args.authorized_by,
        'dry_run': args.dry_run,
        'no_confirm': args.no_confirm,
        'fingerprint': fingerprint_result,
        'action_result': action_result,
        'exit_code': exit_code,
    }

Part B: Add the extended fingerprint for idempotency (~45 min)

The standard fingerprint (v0.1) detects SB6141. The extended fingerprint detects "SB6141 AT DEFAULT CONFIGURATION." Default state markers (firmware-specific; tune per your unit):

  • Default Wi-Fi SSID present (the modem at factory-reset has a known default; the cohort can document the exact value)
  • Default admin-password prompt visible OR absent (depending on firmware version)
  • Default firmware-version banner unchanged (means no firmware-level config diff since reset)
DEFAULT_STATE_MARKERS = [
    re.compile(r'Default\s*SSID', re.IGNORECASE),
    re.compile(r'Default\s*Settings', re.IGNORECASE),
    # Tune per your firmware:
    # re.compile(r'<your unit-specific default marker>', re.IGNORECASE),
]

def fingerprint_with_state(target: str, timeout: float = 5.0) -> dict:
    """Probe + decide both 'is SB6141?' and 'is at default state?'."""
    base = fingerprint(target, timeout=timeout)
    if not base['match']:
        return {**base, 'at_default_state': None}
    # base matched; do a second probe to evaluate state markers
    url = f'http://{target}/'
    try:
        resp = requests.get(url, timeout=timeout)
    except requests.exceptions.RequestException as e:
        return {**base, 'at_default_state': None,
                'state_check_failed': str(e)}
    state_hits = sum(1 for r in DEFAULT_STATE_MARKERS if r.search(resp.text))
    return {**base, 'at_default_state': state_hits >= len(DEFAULT_STATE_MARKERS) // 2}

The state_hits >= len(DEFAULT_STATE_MARKERS) // 2 is a softer match than "all markers present"; firmware variation makes strict-all-markers brittle. Tune per cohort.

Part C: Wire idempotency into main() (~30 min)

Before the destructive action, check if it has already been applied. If yes, exit OK with a "already done" message:

EXIT_ALREADY_DONE = 5  # new exit code

def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    # ... logging setup as in v02 ...

    fp_full = fingerprint_with_state(args.target, timeout=args.timeout)
    if not fp_full['match']:
        log_path = write_run_log(build_run_payload(args, fp_full, None, EXIT_FINGERPRINT_MISMATCH))
        print(f'Error: fingerprint mismatch on {args.target}: {fp_full["reason"]}', file=sys.stderr)
        print(f'(Run log: {log_path})', file=sys.stderr)
        return EXIT_FINGERPRINT_MISMATCH

    print(f'Fingerprint OK: SB6141 at {args.target} (firmware {fp_full.get("firmware_hint", "unknown")})')

    # Idempotency check
    if fp_full.get('at_default_state'):
        print('Target is already at default state; factory-reset is a no-op.')
        print('(Use --rollback PATH or re-configure the modem if a fresh reset is intentional.)')
        log_path = write_run_log(build_run_payload(args, fp_full, {'idempotent_skip': True}, EXIT_ALREADY_DONE))
        print(f'(Run log: {log_path})', file=sys.stderr)
        return EXIT_ALREADY_DONE

    # Dry-run preview
    # ... (as in v02) ...
    if args.dry_run:
        log_path = write_run_log(build_run_payload(args, fp_full, None, EXIT_OK))
        print('Dry-run mode (default). Use --no-dry-run to execute.')
        print(f'(Run log: {log_path})', file=sys.stderr)
        return EXIT_OK

    # ... (confirmation; execute; as in v02; followed by:)
    action_result = do_csrf_reset(args.target, timeout=args.timeout)
    exit_code = EXIT_OK if action_result['success'] else EXIT_RUNTIME_ERROR
    log_path = write_run_log(build_run_payload(args, fp_full, action_result, exit_code))
    print(f'(Run log: {log_path})', file=sys.stderr)
    if action_result['success']:
        print(f'Action complete (HTTP {action_result["status_code"]}). Modem will be reachable again in ~60 seconds.')
        return EXIT_OK
    print(f'Error: destructive request returned HTTP {action_result["status_code"]}', file=sys.stderr)
    return EXIT_RUNTIME_ERROR

Add EXIT_ALREADY_DONE = 5 to the exit-code constants near the top of the file.

Part D: Document the rollback path (~30 min)

Add a --rollback subcommand stub (or argparse mutually-exclusive option):

parser.add_argument(
    '--rollback',
    metavar='RUN_LOG.json',
    default=None,
    help='roll back to the state captured in RUN_LOG.json (pre-action state must have been captured)',
)
parser.add_argument(
    '--capture-pre-state',
    action='store_true',
    default=False,
    help='capture the modem\'s pre-action state and embed it in the run log (required for rollback)',
)

The rollback handler reads the run-log JSON, extracts the pre-state record, and applies the captured configuration via the SB6141 admin endpoints. Stub implementation:

def do_rollback(rollback_path: Path, target: str, authorized_by: str, timeout: float) -> dict:
    """Apply the pre-action state captured in a prior run log.

    This is a stub. The full implementation requires:
      1. Pre-state capture (--capture-pre-state on the destructive run)
      2. Per-config-field setter endpoints documented in the SB6141 admin interface
      3. Per-field POST replay to restore the captured values

    For Lab 7, this stub demonstrates the design pattern; a complete implementation
    is a capstone-stretch exercise.
    """
    with open(rollback_path, 'r') as f:
        prior_run = json.load(f)
    pre_state = prior_run.get('pre_state')
    if not pre_state:
        return {
            'success': False,
            'reason': 'rollback unavailable; prior run did not capture pre-state '
                      '(missing --capture-pre-state on the destructive invocation)',
        }
    # STUB: in a full implementation, replay each pre_state field via the SB6141 admin POST endpoints
    return {
        'success': False,
        'reason': 'rollback implementation stub; pre-state captured but replay not implemented in Lab 7',
        'pre_state_summary': {k: type(v).__name__ for k, v in pre_state.items()},
    }

For Lab 7, the rollback is documented (the design is correct; the design pattern is teachable) but not fully implemented. The capstone (v1.0) may complete the implementation if the cohort has time.

Part E: Test (~30 min)

Run sequence (with authorization line in lab notebook first):

# 1. First dry-run; expect: dry-run preview + run-log file created
python3 -m sb6141_csrf.v03 --target 192.168.100.1 --authorized-by "Alice (cohort 2026-A)"
ls ~/.sb6141-csrf/runs/
cat ~/.sb6141-csrf/runs/<latest>.json  # verify JSON structure

# 2. First actual execution; expect: fingerprint OK, not at default; reset; run-log
python3 -m sb6141_csrf.v03 --target 192.168.100.1 --authorized-by "Alice (cohort 2026-A)" --no-dry-run
# (type "factory-reset" at the prompt)

# 3. Wait ~60s for modem to come back; immediately re-run; expect: at default state; idempotent skip; exit 5
python3 -m sb6141_csrf.v03 --target 192.168.100.1 --authorized-by "Alice (cohort 2026-A)" --no-dry-run
# Expected: "Target is already at default state; factory-reset is a no-op."

# 4. Re-configure the modem manually (change Wi-Fi name); re-run; expect: now NOT at default; ready to reset again
# (interactive; configure via SB6141 admin UI)
python3 -m sb6141_csrf.v03 --target 192.168.100.1 --authorized-by "Alice (cohort 2026-A)" --no-dry-run
# Expected: not at default; proceeds to confirmation prompt

Verify each run produced a JSON file:

ls -la ~/.sb6141-csrf/runs/

Part F: Add unit tests for new behaviors (~20 min)

def test_write_run_log_creates_file(tmp_path, monkeypatch):
    """write_run_log writes a parseable JSON file."""
    monkeypatch.setattr(v03, 'RUNS_DIR', tmp_path)
    payload = {'timestamp': '2026-05-29T00:00:00+00:00', 'tool_version': '0.3', 'exit_code': 0}
    path = v03.write_run_log(payload)
    assert path.exists()
    with open(path) as f:
        data = json.load(f)
    assert data['tool_version'] == '0.3'

def test_idempotency_skips_when_at_default(monkeypatch, tmp_path):
    """When fingerprint_with_state reports at_default_state, the tool exits EXIT_ALREADY_DONE."""
    monkeypatch.setattr(v03, 'RUNS_DIR', tmp_path)
    def fake_fingerprint_with_state(target, timeout=5.0):
        return {'match': True, 'reason': '', 'firmware_hint': '7.5.0', 'at_default_state': True}
    monkeypatch.setattr(v03, 'fingerprint_with_state', fake_fingerprint_with_state)
    called = {'count': 0}
    monkeypatch.setattr(v03, 'do_csrf_reset',
                        lambda target, timeout=10.0: (called.__setitem__('count', called['count'] + 1), {'success': True})[1])
    monkeypatch.setattr('builtins.input', lambda prompt: 'factory-reset')
    rc = v03.main([
        '--target', '1.2.3.4',
        '--authorized-by', 'tester',
        '--no-dry-run',
    ])
    assert rc == v03.EXIT_ALREADY_DONE
    assert called['count'] == 0  # destructive action skipped

Run:

python3 -m pytest tests/ -v
# All tests should pass

Part G: Commit + portfolio (~15 min)

cd ~/adv-101/lab-7
git add sb6141_csrf/v03.py tests/test_v03.py README.md
git commit -m "Lab 7: Tool v0.3 (JSON run logs + idempotency check + rollback stub; tests pass)"

cd ~/adv-101
cat >> lab-portfolio.md <<EOF

### Lab 7 session, 2026-MM-DD HH:MM

**Target:** SB6141 unit serial <X>
**Action:** Tool v0.3 development (JSON run log + idempotency + rollback documentation)
**Authorization basis:** per signed cohort authorization (counter-signed 2026-MM-DD)
**Session duration:** ~3h (3 factory-resets across the session)
**Artifacts produced:** lab-7/sb6141_csrf/v03.py; lab-7/tests/test_v03.py; ~/.sb6141-csrf/runs/*.json
**Incidents:** none; idempotency check correctly identified already-reset state on second consecutive run
EOF
git add lab-portfolio.md
git commit -m "Portfolio: Lab 7 session entry"

Expected output / artifact

~/adv-101/lab-7/ containing:

  • sb6141_csrf/v03.py (Tool v0.3: structured logging + idempotency + rollback stub)
  • tests/test_v03.py (10+ tests; all pass)
  • Updated README.md reflecting v0.3 behaviors

Plus ~/.sb6141-csrf/runs/*.json (one file per run; the audit trail).

What's the failure mode?

  1. Operator deletes the runs directory. Audit trail loss. Defense: the academy can require the cohort to commit ~/.sb6141-csrf/runs/ to the per-cohort Git repository; the academy maintains the canonical copy.
  2. JSON log includes sensitive data. The current schema does not, but a careless future change could. Code review the schema before each tool release.
  3. Idempotency check false-positive. The state markers may report "at default" when the modem is in fact at a configuration that happens to look default. Two-marker discipline (require multiple markers) helps; documenting which markers are checked in the run log preserves verifiability.
  4. Rollback assumes pre-state was captured. The stub correctly fails when pre-state is missing; full implementation in capstone phase.

What would a reviewer ask?

  1. "Show me a run-log entry where the idempotency check fired. Walk me through how it decided 'already done.'"
  2. "Your rollback is a stub. Why a stub now? What would the full implementation require?"
  3. "Where do the JSON logs live? What happens if they fill the disk?"

Stretch (optional)

  1. Implement --capture-pre-state for one configuration field (e.g., Wi-Fi SSID). Reading the value from the SB6141 admin page; embedding in the run log; restoring via a POST request.
  2. Add log rotation. Auto-prune runs older than 90 days; archive instead of delete; document the policy.
  3. JSON-schema-validate the run log. Use the jsonschema library; add a test that every run-log file validates against the schema.
  4. Convert to structlog. Refactor v0.3 to use structlog instead of stdlib logging; compare ergonomics.

Lab 7 v0.1.