labs/Lab-2/Lab-2-Solution/06_Capstone_Auth_Log_Analysis_Solution.ipynb

Module 2 — Lab 6: Capstone — Investigating Suspicious Logins

Topics: Pandas, NumPy, Matplotlib, export, analyst summary

Solution version: This notebook contains the complete solution. Run the cells in order and compare the results with your Lab notebook.

Mission

As a junior security analyst, you must build a complete workflow:

1. load and clean authentication logs;
2. identify IP addresses with repeated failed logins;
3. identify off-hours activity;
4. visualize the results;
5. export the data and write a short conclusion.

In [ ]:

from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

DATA_PATH = Path("data/auth_log.csv")
OUTPUT_DIR = Path("output")
OUTPUT_DIR.mkdir(exist_ok=True)

1. Write a reusable cleaning function

In [ ]:

def clean_auth_log(df):
    result = df.copy().drop_duplicates()
    result["timestamp"] = pd.to_datetime(
        result["timestamp"], errors="coerce", utc=True
    )
    result = result.dropna(subset=["timestamp", "source_ip", "status"])
    result["source_ip"] = result["source_ip"].str.strip()
    result["username"] = result["username"].fillna("unknown").str.strip().str.lower()
    result["status"] = result["status"].str.strip().str.upper()
    result = result[result["status"].isin(["SUCCESS", "FAILED"])]
    result["hour"] = result["timestamp"].dt.hour
    result["is_off_hours"] = (result["hour"] < 6) | (result["hour"] >= 22)
    return result.sort_values("timestamp").reset_index(drop=True)


raw_df = pd.read_csv(DATA_PATH)
clean_df = clean_auth_log(raw_df)
print(f"Raw rows: {len(raw_df)} | Clean rows: {len(clean_df)}")
clean_df.head()

2. Analyze failures and rank IP addresses

In [ ]:

failed_df = clean_df[clean_df["status"] == "FAILED"].copy()
failed_by_ip = (
    failed_df.groupby("source_ip")
    .size()
    .sort_values(ascending=False)
    .rename("failed_attempts")
)

THRESHOLD = 3
suspicious_ips = failed_by_ip[failed_by_ip >= THRESHOLD]

print("Failed attempts by IP:")
print(failed_by_ip)
print("\nSuspicious IPs:")
print(suspicious_ips)

3. Calculate additional investigation metrics

In [ ]:

status_counts = clean_df["status"].value_counts().reindex(["SUCCESS", "FAILED"], fill_value=0)
off_hours_failed = failed_df[failed_df["is_off_hours"]]
top_ip = failed_by_ip.index[0]
top_ip_count = int(failed_by_ip.iloc[0])
failure_rate = np.round((status_counts["FAILED"] / status_counts.sum()) * 100, 2)

print(f"Top suspicious IP: {top_ip} ({top_ip_count} failures)")
print(f"Failure rate: {failure_rate}%")
print(f"Off-hours failed attempts: {len(off_hours_failed)}")

4. Visualize the findings

In [ ]:

fig, axes = plt.subplots(1, 2, figsize=(11, 4))

status_counts.plot(
    kind="bar",
    ax=axes[0],
    color=["#2a9d8f", "#e76f51"],
    title="Authentication status",
)
axes[0].set_xlabel("Status")
axes[0].set_ylabel("Number of attempts")
axes[0].tick_params(axis="x", rotation=0)

failed_by_ip.head(5).sort_values().plot(
    kind="barh",
    ax=axes[1],
    color="#e76f51",
    title="Top IPs by failed logins",
)
axes[1].set_xlabel("Failed attempts")
axes[1].set_ylabel("Source IP")

fig.suptitle("Authentication Log Investigation")
fig.tight_layout()
fig.savefig(OUTPUT_DIR / "auth_log_analysis.png", dpi=150, bbox_inches="tight")
plt.show()

5. Export results and write a short report

In [ ]:

clean_df.to_csv(OUTPUT_DIR / "capstone_clean_auth_log.csv", index=False)
suspicious_ips.to_csv(
    OUTPUT_DIR / "capstone_suspicious_ips.csv",
    header=True,
)

summary = f'''SECURITY ANALYSIS SUMMARY
=========================
Clean events: {len(clean_df)}
Failed events: {int(status_counts['FAILED'])}
Failure rate: {failure_rate}%
Top suspicious IP: {top_ip} ({top_ip_count} failed attempts)
Off-hours failed attempts: {len(off_hours_failed)}
Decision threshold: {THRESHOLD} failed attempts

Recommended action:
- Review events from {top_ip}.
- Correlate off-hours failures with endpoint and identity logs.
- Validate whether the account owners recognize these attempts.
'''

(OUTPUT_DIR / "capstone_summary.txt").write_text(summary, encoding="utf-8")
print(summary)

Self-check

In [ ]:

assert len(raw_df) > len(clean_df)
assert clean_df.duplicated().sum() == 0
assert failed_by_ip.index[0] == "198.51.100.14"
assert top_ip_count >= THRESHOLD
assert len(off_hours_failed) >= 1
assert (OUTPUT_DIR / "auth_log_analysis.png").exists()
assert (OUTPUT_DIR / "capstone_summary.txt").exists()
print("✓ Capstone lab completed")

Report questions

1. Which IP address should be investigated first, and what evidence supports that decision?
2. In what situations could a threshold of three failed attempts produce a false positive?
3. Which additional log sources should be reviewed before deciding to block an IP address?