LEARN. PRACTICE. GROW.Prepare decisions before the overnight incident
In a fictional funds service, a credential performs unusual exports shortly before closing. The alert is an observation for triage; it does not alone identify a person or the full incident scope. Define who declares the incident, authorizes containment, coordinates the provider and communicates with business owners. Prepare controlled emergency access, alternate contacts and escalation criteria before needing them. NIST SP800-61 Rev.3, finalized in April 2025, integrates incident response into risk management and supersedes Rev.2. Use it to structure decisions and responsibilities without turning the document into bank-specific rules. The operational objective is to limit impact while recording what is known, assumed and still unconfirmed.
Preserve the original and document acquisition
The lab creates a fictional JSONL file, copies its bytes into a local acquisition and records declared origin, size, SHA-256, collector and copy time. Analysis uses that copy and checks the original remains unchanged. This demonstrates concrete local operations, not forensic acquisition from a compromised system. A matching hash supports equality with the stored reference; it does not prove events were true before collection. The exercise changes the file and then shows that recalculating an equally editable manifest hides the mismatch. In actual operations, preserve provenance, access, transfers and custody procedures appropriate to the case. Avoid posting full sensitive data into an incident channel containing unauthorized participants.
Distinguish event time, receipt and causal order
In the fictional data, 11:00 with offset +01:00 equals 10:00 UTC. Credential use reaches the collector only at 10:04, after an application response received at 10:01:10. Sorting by receipt produces a different narrative from sorting by declared event time. An agent declares 10:06 for a record received at 10:02: calculated delay is negative four minutes, requiring clock or metadata investigation. Do not rewrite the original to make the sequence convenient. The lab rejects timestamps without a timezone and flags the anomaly. Even after UTC normalization, overlapping error bounds prevent asserting strict order. Request IDs and temporal proximity help investigation; alone they establish neither causality nor human attribution.
Correlate without erasing contradictions
There are six raw records and five distinct events in the example. One event is delivered twice with different receipt times. The lab groups by account, source and eventId while preserving both records in the original. The same eventId in another account is not discarded. One variant changes the action of an event with the same identity: the analyzer flags a conflict, retains the first for inspection and requires reviewing the rejected line before drawing conclusions. It does not treat the contradiction as a benign duplicate. Thirty checks passed in two local runs. No SIEM, actual compromise or CloudTrail validation occurred. In CloudTrail, enabling digest delivery does not automatically execute integrity validation; documentation distinguishes those operations.
Containment does not close the incident
Containment should consider impact, speed, permissions and opportunities to preserve evidence. If damage is ongoing, do not postpone every response action just to achieve perfect collection; coordinate authorized actions and record the tradeoff. Isolating one instance does not establish that copied credentials stopped working elsewhere. Restoring an endpoint also does not prove persistence and exploited causes were removed. Define return-to-service criteria, functional validation, heightened monitoring and improvement actions. Communicate confirmed facts and uncertainties with a scheduled next update. Notification duties depend on context and should be assessed by appropriate owners; this lesson invents no universal deadline. Summary: preserve, correlate with limits, contain under authority and validate recovery.
"""Original local incident-analysis exercise. All events and clock bounds are fictional."""
import datetime as dt
import hashlib
import json
from pathlib import Path
import shutil
import sys
import tempfile
def digest(data):
return hashlib.sha256(data).hexdigest
def timestamp(value):
if not isinstance(value, str):
raise ValueError('timestamp type')
parsed = dt.datetime.fromisoformat(value.replace('Z', '+00:00'))
if parsed.tzinfo is None or parsed.utcoffset is None:
raise ValueError('timezone required')
return parsed.astimezone(dt.timezone.utc)
def analyze(data):
unique, duplicates, rejected = {}, [], []
for lineno, line in enumerate(data.splitlines, 1):
try:
row = json.loads(line)
if not isinstance(row, dict):
raise ValueError('object required')
for field in ('account', 'source', 'eventId', 'requestId', 'action'):
if not isinstance(row.get(field), str) or not row[field]:
raise ValueError('missing identity field')
event = timestamp(row['eventTime'])
received = timestamp(row['receivedTime'])
key = (row['account'], row['source'], row['eventId'])
# Preserve receive timestamps separately; compare producer event fields.
event_fields = {k: v for k, v in row.items if k!= 'receivedTime'}
canonical = json.dumps(event_fields, sort_keys=True, separators=(',', ':'))
if key in unique:
if canonical!= unique[key]['canonical']:
raise ValueError('conflicting event identity')
duplicates.append({'line': lineno, 'identity': list(key), 'receivedTime': row['receivedTime']})
continue
unique[key] = {'row': row, 'line': lineno, 'canonical': canonical,
'eventUTC': event.isoformat, 'receivedUTC': received.isoformat,
'delaySeconds': (received-event).total_seconds,
'clockWarning': received < event}
except (ValueError, KeyError, TypeError) as error:
rejected.append({'line': lineno, 'reason': str(error)})
ordered = sorted(unique.values, key=lambda x: (x['eventUTC'], x['row']['account'], x['row']['source'], x['row']['eventId']))
# This deterministic display is not proof of a causal ordering.
return {'events': ordered, 'duplicates': duplicates, 'rejected': rejected}
def definitely_before(left, right, left_error_seconds, right_error_seconds):
if left_error_seconds < 0 or right_error_seconds < 0:
raise ValueError('negative clock bound')
return timestamp(left)+dt.timedelta(seconds=left_error_seconds) < timestamp(right)-dt.timedelta(seconds=right_error_seconds)
def event(account, source, eid, event_time, received, action, request='r-17'):
return {'account': account, 'source': source, 'eventId': eid, 'requestId': request,
'eventTime': event_time, 'receivedTime': received, 'action': action}
def encode(rows):
return ('\n'.join(json.dumps(row, ensure_ascii=False, separators=(',', ':')) for row in rows)+'\n').encode
def main(output=None):
checks = []
def check(name, condition):
if not condition:
raise AssertionError(name)
checks.append({'id': name, 'pass': True})
rows = [
event('funds-A', 'identity', 'auth-1', '2026-10-07T11:00:00+01:00', '2026-10-07T10:04:00Z', 'credential-used'),
event('funds-A', 'gateway', 'export-1', '2026-10-07T10:01:00Z', '2026-10-07T10:02:00Z', 'export-requested'),
event('funds-A', 'application', 'response-1', '2026-10-07T10:01:02Z', '2026-10-07T10:01:10Z', 'response-accepted'),
event('funds-A', 'gateway', 'export-1', '2026-10-07T10:01:00Z', '2026-10-07T10:05:00Z', 'export-requested'),
event('funds-B', 'gateway', 'export-1', '2026-10-07T10:01:01Z', '2026-10-07T10:02:05Z', 'export-requested'),
event('funds-A', 'agent', 'clock-1', '2026-10-07T10:06:00Z', '2026-10-07T10:02:00Z', 'agent-observation'),
]
with tempfile.TemporaryDirectory(prefix='dr-incident-local-') as temp:
root = Path(temp);original = root/'synthetic-events.jsonl'raw = encode(rows);original.write_bytes(raw)
acquired = root/'acquired.jsonl'shutil.copyfile(original, acquired)
manifest = {'sha256': digest(raw), 'sizeBytes': len(raw), 'source': 'generated synthetic fixture', 'collector': 'local exercise', 'acquiredAt': dt.datetime.now(dt.timezone.utc).isoformat}
check('acquired-bytes-match-original', acquired.read_bytes == original.read_bytes)
check('acquired-hash-matches-manifest', digest(acquired.read_bytes) == manifest['sha256'])
check('acquired-size-matches-manifest', acquired.stat.st_size == manifest['sizeBytes'])
check('timezone-normalization-same-instant', timestamp(rows[0]['eventTime']) == timestamp('2026-10-07T10:00:00Z'))
result = analyze(acquired.read_bytes)
check('six-raw-records-preserved', len(acquired.read_bytes.splitlines) == 6)
check('five-distinct-events', len(result['events']) == 5)
check('one-received-duplicate', len(result['duplicates']) == 1)
check('valid-fixture-no-rejected-lines', not result['rejected'])
check('same-event-id-other-account-retained', sum(e['row']['eventId']=='export-1' for e in result['events']) == 2)
check('duplicate-receive-time-preserved', result['duplicates'][0]['receivedTime'] == '2026-10-07T10:05:00Z')
check('event-time-starts-with-credential-use', result['events'][0]['row']['eventId'] == 'auth-1')
arrival = sorted(result['events'], key=lambda e: e['receivedUTC'])
check('arrival-order-starts-with-response', arrival[0]['row']['eventId'] == 'response-1')
check('arrival-order-differs-from-event-time', [e['row']['eventId'] for e in arrival]!= [e['row']['eventId'] for e in result['events']])
clock = next(e for e in result['events'] if e['row']['eventId']=='clock-1')
check('future-event-negative-delay', clock['delaySeconds'] == -240)
check('future-event-clock-warning', clock['clockWarning'])
naive = event('funds-A','gateway','bad-time','2026-10-07T10:00:00','2026-10-07T10:01:00Z','unknown')
check('naive-timestamp-rejected', analyze(encode([naive]))['rejected'][0]['reason'] == 'timezone required')
broken = analyze(raw+b'{broken json}\n')
check('malformed-line-is-quarantined', len(broken['rejected']) == 1 and len(broken['events']) == 5)
missing = dict(rows[0]);del missing['account']
check('non-object-json-rejected', analyze(b'[]\n')['rejected'][0]['reason'] == 'object required')
check('missing-account-rejected', analyze(encode([missing]))['rejected'][0]['reason'] == 'missing identity field')
conflicting = dict(rows[1]);conflicting['action']='export-denied'
conflict_result = analyze(encode(rows+[conflicting]))
check('conflicting-identity-flagged', conflict_result['rejected'][0]['reason'] == 'conflicting event identity')
check('conflict-not-counted-as-benign-duplicate', len(conflict_result['duplicates']) == 1)
try:
definitely_before('2026-10-07T10:00:00Z','2026-10-07T10:01:00Z',-1,2)
rejected_negative = False
except ValueError:
rejected_negative = True
check('negative-clock-bound-rejected', rejected_negative)
check('narrow-clock-bounds-order-known', definitely_before('2026-10-07T10:00:00Z','2026-10-07T10:01:00Z',2,2))
check('overlapping-clock-bounds-order-unknown', not definitely_before('2026-10-07T10:00:00Z','2026-10-07T10:01:00Z',90,90))
check('touching-bounds-not-strictly-before', not definitely_before('2026-10-07T10:00:00Z','2026-10-07T10:01:00Z',30,30))
a_only = [e for e in result['events'] if e['row']['account']=='funds-A' and e['row']['requestId']=='r-17']
check('correlation-keeps-account-boundary', len(a_only) == 4 and all(e['row']['account']=='funds-A' for e in a_only))
edited = raw.replace(b'export-requested', b'export-denied', 1)
check('edited-bytes-fail-original-hash', digest(edited)!= manifest['sha256'])
forged_manifest = {'sha256': digest(edited)}
check('rewriting-both-file-and-manifest-evades-untrusted-hash', digest(edited) == forged_manifest['sha256'])
check('original-remains-unchanged-after-analysis', original.read_bytes == raw and acquired.read_bytes == raw)
facts = {'rawRecords': 6, 'uniqueEvents': 5, 'duplicateRecords': 1, 'clockWarnings': 1,
'firstByEventTime': result['events'][0]['row']['eventId'], 'firstByReceiveTime': arrival[0]['row']['eventId'],
'futureEventDelaySeconds': clock['delaySeconds'], 'rawSHA256': manifest['sha256'], 'sizeBytes': len(raw)}
check('owned-temporary-files-cleaned', not root.exists)
report = {'executedAt': dt.datetime.now(dt.timezone.utc).isoformat, 'runtime': sys.version.split[0],
'scriptSHA256': digest(Path(__file__).read_bytes), 'passed': len(checks), 'failed': 0,
'checks': checks, 'facts': facts, 'acquisitionManifest': manifest,
'scope': 'Actual local file copy, SHA-256 checks and parsing of fictional JSONL incident events.',
'limits': ['No actual compromise, cloud API, CloudTrail validation, SIEM or forensic imaging',
'Synthetic event timestamps and clock bounds, not measured clocks',
'A local mutable manifest is not an authenticated source or tamper-proof custody record',
'Display order and matching request IDs do not prove causality or human attribution',
'First conflicting event retained for inspection; rejected conflicts must be reviewed before conclusions',
'No legal admissibility or notification-deadline determination',
'No independent specialist review']}
if output:Path(output).write_text(json.dumps(report,indent=2)+'\n')
print(json.dumps({'passed': report['passed'], 'failed': 0, 'scriptSHA256': report['scriptSHA256'], 'facts': facts}))
if __name__ == '__main__':main(sys.argv[1] if len(sys.argv)>1 else None)
11:00+01:00 equals 10:00Z; receipt order does not prove causal order. Six retained lines contain five distinct events.
Common pitfalls
Confusing hashes with authenticity; receipt with occurrence; equal IDs across accounts with duplicates; a restored endpoint with a closed incident.
Related topics: Operations, risk and change management · Evidence, responsibility and acceptance
Build a timeline with explicit limits and coordinate containment, evidence and recovery.
Reference: Incident Response Recommendations and Considerations for Cybersecurity Risk Management · CCSP examination outline effective 2026-08-01; January2026 V2 PDF