file_integrity_monitor.py

"""
A simple file integrity monitor — companion script for the post:
  "Python Quick Guide: A Simple File Integrity Monitor"

The baseline/compare functions are exactly as they appear in the post. The demo
creates a sample tree, builds a baseline, then adds, modifies, and deletes a
file before re-running the diff so the report prints one of each change class —
that [ADDED]/[MODIFIED]/[DELETED] block is the screenshot for the post.

Run:
    python file_integrity_monitor.py
"""

import hashlib
import os
import json
import shutil


# --- post code: Step 1 -------------------------------------------------------
def sha256_file(path, chunk_size=65536):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            h.update(chunk)
    return h.hexdigest()


def build_baseline(root):
    baseline = {}
    for dirpath, _dirs, files in os.walk(root):
        for name in files:
            full = os.path.join(dirpath, name)
            try:
                baseline[full] = sha256_file(full)
            except (PermissionError, FileNotFoundError, OSError):
                continue   # skip what we cannot read
    return baseline


# --- post code: Step 2 -------------------------------------------------------
def save_baseline(baseline, path):
    with open(path, "w", encoding="utf-8") as f:
        json.dump(baseline, f, indent=2)


def load_baseline(path):
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


# --- post code: Step 3 -------------------------------------------------------
def compare(baseline, root):
    current = build_baseline(root)
    base_paths = set(baseline)
    curr_paths = set(current)

    added = curr_paths - base_paths
    deleted = base_paths - curr_paths
    modified = {p for p in base_paths & curr_paths if baseline[p] != current[p]}

    return {"added": sorted(added),
            "deleted": sorted(deleted),
            "modified": sorted(modified)}


# --- post code: Step 4 -------------------------------------------------------
def run(root, baseline_path):
    if not os.path.exists(baseline_path):
        save_baseline(build_baseline(root), baseline_path)
        print(f"Baseline created for {root}")
        return

    changes = compare(load_baseline(baseline_path), root)
    if not any(changes.values()):
        return   # silent when nothing changed

    for category, paths in changes.items():
        for p in paths:
            print(f"[{category.upper()}] {p}")


# --- demo harness (not in the post) -----------------------------------------
def _write(path, data):
    with open(path, "w", encoding="utf-8") as f:
        f.write(data)


if __name__ == "__main__":
    base_dir = os.path.dirname(os.path.abspath(__file__))
    watched = os.path.join(base_dir, "_demo_fim", "wwwroot")
    baseline_path = os.path.join(base_dir, "_demo_fim", "baseline.json")

    # fresh sample web root
    if os.path.exists(os.path.dirname(watched)):
        shutil.rmtree(os.path.dirname(watched))
    os.makedirs(watched)
    _write(os.path.join(watched, "index.html"), "<h1>home</h1>")
    _write(os.path.join(watched, "style.css"), "body{margin:0}")
    _write(os.path.join(watched, "app.js"), "console.log('v1')")

    # first run: establish the known-good baseline
    run(watched, baseline_path)

    # simulate drift: modify one file, add one, delete one.
    # (the added file uses benign content on purpose — a literal webshell
    # string like system($_GET[...]) gets quarantined by Defender before the
    # monitor can hash it, which would hide the [ADDED] line.)
    _write(os.path.join(watched, "app.js"), "console.log('TAMPERED')")
    _write(os.path.join(watched, "unexpected.html"), "<h1>not deployed by me</h1>")
    os.remove(os.path.join(watched, "style.css"))

    # second run: the diff that matters
    print("\nChange report:")
    run(watched, baseline_path)