Security Notes
Coding

Security Coding Round — Eight Interview Problems

Security engineering loops almost always include a coding round, and it is rarely a puzzle about graphs. It is usually "here's a log, find the bad thing": count failed logins, spot brute force or spraying, flag impossible travel, parse CloudTrail, pull indicators out of a report. These eight problems cover the patterns behind most of those questions. Every solution below is copied from security_exercises.py, which tests itself: run python3 python/security_exercises.py.

10 min read 9 sections 3 model answers
Jump to questions Drill these
Memory hook

say it before you type itInterviewers grade the conversation as much as the code. For every problem: ask two clarifying questions, state the approach and its complexity in one sentence, write the simplest correct version, then volunteer the production follow-ups (scale, false positives, memory). The general patterns (sliding window, heaps, hash maps) are in the patterns primer.

How to approach any log question

  1. Pin down the inputClarify2 min
    • Format, size, sorted or not, what counts as an event, what output is wanted
  2. Name the data structurePlan1 min
    • Counter for "top N", deque for "within T seconds", dict of sets for "distinct", last-seen dict for "consecutive"
  3. Simplest correct versionCode15 min
    • Stream the input, compile regexes once, use the standard library
  4. Run it on an example you make upTest3 min
    • Include an edge case: empty input, window boundary, malformed line
  5. Production follow-upsExtend5 min
    • Bigger data, unsorted data, false positives, where state lives

1. Top attacking IPs in an auth log

The askGiven /var/log/auth.log (could be many gigabytes), print the ten IPs with the most failed SSH logins.

Clarify firstDoes invalid user count? (Yes: it's still a failed attempt.) IPv6 too? Is the file rotated or compressed?

ApproachStream line by line (never read() a huge file), match with one compiled regex, count with Counter, finish with most_common.

python
FAILED_SSH = re.compile(r"Failed password for (?:invalid user )?(\S+) from (\d{1,3}(?:\.\d{1,3}){3})")

def top_failed_ips(lines, n=10):
    counts = Counter()
    for line in lines:  # streams: works on a 50 GB file opened lazily
        m = FAILED_SSH.search(line)
        if m:
            counts[m.group(2)] += 1
    return counts.most_common(n)

ComplexityO(n) time over lines, O(k) memory for k distinct IPs. most_common(10) uses a heap, so it doesn't sort everything.

Follow-ups interviewers askMake it work for gzip files (gzip.open(path, 'rt')). Count usernames tried per IP too. What if it's 1 TB across 500 hosts? Count per file in parallel, then merge the Counters (map-reduce).

2. Brute force in a time window

The askFlag any IP with at least 5 failed logins within 60 seconds.

Clarify firstAre events sorted by time? Sliding window or fixed buckets? (Sliding: fixed one-minute buckets miss 3 failures at :59 and 3 at :01.)

ApproachOne deque of timestamps per IP. Append the new event, pop from the left everything older than the window, and check the length.

python
def brute_force(events, threshold=5, window=60):
    """events: (epoch_seconds, ip) in time order. Returns IPs with >= threshold failures inside any window."""
    recent = defaultdict(deque)
    flagged = set()
    for ts, ip in events:
        q = recent[ip]
        q.append(ts)
        while q[0] <= ts - window:  # drop failures that fell out of the window
            q.popleft()
        if len(q) >= threshold:
            flagged.add(ip)
    return flagged

ComplexityO(n) amortised: each timestamp is appended and popped once. Memory O(active IPs × threshold).

Follow-ups interviewers askMemory grows with idle IPs: evict deques that go empty. Unsorted input: sort first, or allow a small lateness buffer. Distributed: key a stream processor (Kafka, Flink) by IP.

3. Password spraying

The askSpraying is one source trying a common password against many accounts, staying under per-account lockout. Flag sources failing against 10 or more distinct users in 5 minutes.

Clarify firstIs the source an IP, or should we group by user agent or ASN? Do we have successes too? (A success after a spray is the real emergency.)

ApproachSame sliding window as brute force, but keep (time, user) pairs and count distinct users instead of events.

python
def password_spray(events, min_users=10, window=300):
    """events: (epoch_seconds, ip, user) failures in time order. Spray = one source, many different users."""
    recent = defaultdict(deque)
    flagged = {}
    for ts, ip, user in events:
        q = recent[ip]
        q.append((ts, user))
        while q[0][0] <= ts - window:
            q.popleft()
        users = {u for _, u in q}
        if len(users) >= min_users:
            flagged[ip] = max(flagged.get(ip, 0), len(users))
    return flagged

ComplexityO(n × w) as written, because it rebuilds the set each event (w = events in the window). For high volume keep a Counter of users inside the window and update it on append and pop, making it O(n).

Follow-ups interviewers askHow would you catch a spray spread over 1,000 IPs? Group by target users and time instead: many users each failing once within minutes, from many sources.

4. Impossible travel

The askFlag a user whose consecutive logins are further apart than an airliner could fly in the time between them.

Clarify firstDo we have coordinates or only IPs? (Geolocate IPs first; accuracy is city-level at best.) Are VPN or corporate egress IPs known?

ApproachRemember each user's last login. For each new one compute great-circle distance with the haversine formula and the implied speed.

python
def haversine_km(a, b):
    lat1, lon1, lat2, lon2 = map(math.radians, (*a, *b))
    h = math.sin((lat2 - lat1) / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin((lon2 - lon1) / 2) ** 2
    return 2 * 6371 * math.asin(math.sqrt(h))

def impossible_travel(logins, max_kmh=900):
    """logins: (epoch_seconds, user, (lat, lon)) in time order. Flags consecutive logins faster than an airliner."""
    last = {}
    alerts = []
    for ts, user, loc in logins:
        if user in last:
            pts, ploc = last[user]
            km = haversine_km(ploc, loc)
            hours = max((ts - pts) / 3600, 1 / 60)  # floor at one minute so same-second logins don't divide by zero
            if km > 100 and km / hours > max_kmh:  # ignore short hops: IP geolocation is only city-accurate
                alerts.append((user, round(km), round(km / hours)))
        last[user] = (ts, loc)
    return alerts

ComplexityO(n) time, O(users) memory.

Follow-ups interviewers askFalse positives: VPNs, mobile carriers, cloud egress. Suppress known corporate ranges and require distance > 100 km. Better signal: impossible travel plus a new device or a sign-in from an anonymising proxy.

5. Suspicious CloudTrail activity

The askGiven CloudTrail events as JSON, alert on defence-evasion and persistence calls, and on an identity using an API it has never used before.

Clarify firstWhich APIs are sensitive for us? Is there a baseline period, or start learning now? Is this real-time or a batch hunt?

ApproachA set of sensitive event names for instant alerts, plus a per-identity set of APIs seen so far for first-use alerts.

python
SENSITIVE = {"StopLogging", "DeleteTrail", "UpdateTrail", "PutBucketPolicy", "DeleteBucketPolicy",
             "CreateAccessKey", "AttachUserPolicy", "PutUserPolicy", "DisableKey", "ScheduleKeyDeletion"}

def cloudtrail_alerts(records):
    """records: CloudTrail events as dicts. Flags sensitive calls and the first time an identity uses any API."""
    seen = defaultdict(set)
    alerts = []
    for r in sorted(records, key=lambda r: r["eventTime"]):
        who = r.get("userIdentity", {}).get("arn", "unknown")
        api = r["eventName"]
        if api in SENSITIVE:
            alerts.append(("sensitive", who, api, r.get("sourceIPAddress")))
        elif seen[who] and api not in seen[who]:
            alerts.append(("first-use", who, api, r.get("sourceIPAddress")))
        seen[who].add(api)
    return alerts

ComplexityO(n log n) for the sort, O(identities × APIs) memory.

Follow-ups interviewers askFirst-use is noisy for a new identity, so skip identities younger than a learning period. Where does the baseline live across runs? (A key-value store, or a SIEM lookup table.) Which fields would you add? errorCode (AccessDenied bursts mean reconnaissance), userAgent, sourceIPAddress.

6. Extract indicators of compromise from a report

The askPull IPv4 addresses, SHA-256 hashes and domains out of a threat-intel report that writes them defanged (hxxp://evil[.]com).

Clarify firstWhich indicator types? Should private IP ranges be dropped? Which top-level domains?

ApproachRefang first, then run one regex per type. Validate octets in the IP regex (999.1.1.1 is not an address), and deduplicate with sets.

python
IPV4 = re.compile(r"\b(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)\b")
SHA256 = re.compile(r"\b[a-fA-F0-9]{64}\b")
DOMAIN = re.compile(r"\b(?:[a-z0-9-]+\.)+(?:com|net|org|io|ru|cn|info|xyz|top)\b", re.I)

def refang(text):
    return (text.replace("hxxp", "http").replace("[.]", ".").replace("(.)", ".").replace("[:]", ":"))

def extract_iocs(text):
    t = refang(text)
    return {"ipv4": sorted(set(IPV4.findall(t))), "sha256": sorted({h.lower() for h in SHA256.findall(t)}),
            "domains": sorted({d.lower() for d in DOMAIN.findall(t)})}

ComplexityO(length of text).

Follow-ups interviewers askWhy not one giant regex? (Unreadable and slow to fix.) Domain regexes catch file names like report.zip: check against a top-level-domain list. Drop private and documentation ranges with the ipaddress module.

7. Triage a JWT

The askDuring an incident you find a JWT in a log. Decode it and list anything suspicious, without a library.

Clarify firstDo we have the signing key? (If not, we can only inspect, never trust.) Which algorithms does the service expect?

ApproachA JWT is three base64url parts. Restore the stripped = padding, decode the header and payload, and check the classic red flags.

python
def jwt_findings(token, now=None):
    """Decode WITHOUT verifying (triage only) and list red flags."""
    def b64(part):
        return json.loads(base64.urlsafe_b64decode(part + "=" * (-len(part) % 4)))
    header, payload = (b64(p) for p in token.split(".")[:2])
    now = now or datetime.now(timezone.utc).timestamp()
    issues = []
    if header.get("alg", "").lower() == "none":
        issues.append("alg=none: unsigned token")
    if {"jku", "jwk", "x5u"} & header.keys():
        issues.append("header names its own key: verify only against a pinned key")
    if "exp" not in payload:
        issues.append("no expiry")
    elif payload["exp"] < now:
        issues.append("expired")
    if payload.get("exp", now) - payload.get("iat", now) > 86400:
        issues.append("lifetime over 24h")
    return issues

ComplexityO(token length).

Follow-ups interviewers askWhy must a server never pick the algorithm from the token header? (alg: none and the RS256-to-HS256 confusion attack.) Why is decoding fine here but dangerous in an authorization path? (Decoding is not verifying.)

8. Detect beaconing

The askMalware often checks in with its command-and-control server at a fixed interval. From connection logs, find source and destination pairs that talk with suspiciously regular timing.

Clarify firstHow many connections make a pattern? How much jitter do we tolerate? (Many families add random jitter of 10–20%.)

ApproachGroup timestamps by (source, destination), compute the gaps between consecutive connections, and use the coefficient of variation (standard deviation ÷ mean). Near 0 means clockwork.

python
def beacons(conns, min_events=6, max_jitter=0.1):
    """conns: (epoch_seconds, src, dst) in time order. Flags pairs with very regular intervals (malware check-ins)."""
    times = defaultdict(list)
    for ts, src, dst in conns:
        times[(src, dst)].append(ts)
    found = []
    for pair, ts in times.items():
        if len(ts) < min_events:
            continue
        gaps = [b - a for a, b in zip(ts, ts[1:])]
        m = mean(gaps)
        if m > 0 and pstdev(gaps) / m <= max_jitter:  # coefficient of variation: 0 = perfectly regular
            found.append((pair, round(m)))
    return found

ComplexityO(n) time, O(n) memory for the timestamps.

Follow-ups interviewers askLegitimate beacons exist: update checks, NTP, monitoring agents. Filter by destination popularity (how many internal hosts talk to it) and domain age. Malware with high jitter defeats this, so look at the gap distribution, not just its spread.

Interview Questions

Q
Why stream a log file line by line instead of reading it all?
Model answer

A production auth log can be tens of gigabytes, and read() or readlines() loads it all into memory and can crash the process or the box. Iterating over the file object reads one line at a time, so memory stays constant and only the state I keep, like a Counter of IPs, grows. It also lets the same code read from gzip.open or standard input in a pipeline, which is how these scripts run in practice.

Q
How would you detect brute force across many log files on many hosts?
Model answer

For a one-off hunt, I'd run the counting step per file in parallel and merge the partial counters, which is map-reduce. For ongoing detection I wouldn't run scripts at all: ship the logs to a SIEM or a stream processor and express it as a windowed aggregation keyed by source IP, like "count of failures by IP over 60 seconds, at least 5". The Python version is still worth writing in an interview, because it shows the sliding-window logic the SIEM query hides.

Q
Brute force versus password spraying: how does your detection differ?
Model answer

Brute force is many attempts against one account, so I count failures per source or per account within a window, and account lockout also catches it. Spraying is one or two common passwords against many accounts, deliberately staying under lockout thresholds, so I count distinct usernames per source in the window. Distributed spraying from many IPs needs a third view: many different users each failing once within minutes. And any success from a spraying source is the real alert, because that's a compromised account.