#!/usr/bin/env python3
"""
OpsLab anonymization tool.
Replaces service, host/node, and people/department names with stable
pseudonyms (SVC-XXXXXX, NODE-XXXXXX, PERS-XXXXXX) BEFORE you send an
incident log to OpsLab. Reads .csv, .txt, .json, .xlsx; always writes
plain UTF-8 CSV.
Requires Python 3.8+ already installed on the machine you run it on —
that is the one thing this script cannot supply itself. Beyond that:
standard library only, no pip install, no other dependency, ever.
What it changes:
- values in columns that look like service / host / assignee / department
- the same names when they appear inside free-text description columns
- IP addresses, hostnames, and emails inside free-text columns (regex mask)
What it never changes:
- column headers, dates, numbers, IDs, severity/priority/status values
With no arguments, this opens a file picker instead of a terminal — choose
the log file, choose where to save the anonymized copy, done. That only
works if double-clicking a .py file on your machine is set up to run it
with python.exe rather than open it in a text editor; if it opens as text
instead, use the terminal instead (see below) — the .py file itself is
fine either way, only how the OS launches it differs. At the end the
picker flow offers to open the OpsLab upload page in your browser; that's
the only network activity in the whole script, it's optional, and it never
sends the file itself — you still pick and attach it yourself.
Usage from a terminal (always works once Python is installed):
python opslab-anonymize.py incidents.xlsx
python opslab-anonymize.py incidents.xlsx --dict incidents.dict.csv
Output (next to the input file, unless -o / --dict override the path):
incidents.anon.csv -> send this file to OpsLab
incidents.dict.csv -> keep this yourself. Needed to:
(a) re-run on a later export with the SAME
pseudonyms, so two audits stay comparable
(b) look a pseudonym back up in the report
Known limitation: if a date column in an .xlsx file is stored as a native
Excel date (not text), this reader returns the raw numeric serial, not a
date string, because converting it correctly requires the workbook's number
formats (out of scope for a zero-dependency script). Most ITSM exports
already store dates as text and are unaffected. If your report's date
column looks like "45123.5" after anonymizing, re-export as CSV from Excel
instead of sending the .xlsx directly.
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import io
import json
import os
import re
import secrets
import sys
import xml.etree.ElementTree as ET
import zipfile
from typing import Optional
# ---------------------------------------------------------------------------
# Column-name heuristics. Mirrors the keyword lists in the OpsLab parser
# (app/services/stats_engine.py, app/services/parser.py) but is kept as a
# plain literal here so this script has zero dependencies on the rest of
# the codebase and can run on a bare Python install on the client's machine.
# ---------------------------------------------------------------------------
_SERVICE_HINTS = {
"service", "component", "module", "app", "application", "source",
"logger", "category", "system",
"сервис", "компонент", "система", "приложение", "услуга",
"servis", "komponenta", "aplikacija", "sistem",
}
_HOST_HINTS = {
"host", "hostname", "server", "node", "machine", "device",
"узел", "сервер", "хост", "машина",
"čvor", "cvor",
}
_PERSON_HINTS = {
"assignee", "owner", "responsible", "department", "team", "group", "engineer",
"исполнитель", "ответственный", "подразделение", "отдел", "группа", "команда",
"izvršilac", "izvrsilac", "odgovoran", "odeljenje", "tim",
}
# Hard exclusion, checked BEFORE the hints above and by whole-token equality
# (never substring): a short hint like "tim" (Serbian "team") is a substring
# of the English word "time", so plain substring matching would misclassify
# a "start_time" / "end_time" column as a person column and destroy the
# dates. Dates, durations, IDs, and status/priority/severity are the columns
# this script promises never to touch — this list is the enforcement of
# that promise, independent of whatever the service/host/person hints above
# happen to contain.
_EXCLUDE_TOKENS = {
"start", "end", "begin", "opened", "closed", "created", "resolved",
"occurred", "logged", "date", "time", "timestamp", "datetime", "ts",
"duration", "id", "ticket", "number", "priority", "severity", "status",
"level",
"дата", "время", "длительность", "начало", "окончание", "создан",
"закрыт", "открыт", "решен", "приоритет", "статус", "уровень", "длит",
"datum", "vreme", "trajanje", "početak", "pocetak", "kraj", "otvoren",
"zatvoren", "prioritet", "nivo",
}
_DESCRIPTION_HINTS = {
"description", "descr", "desc", "notes", "note", "root_cause", "rootcause",
"root-cause", "message", "msg", "text", "body", "detail", "details",
"comment", "comments", "info", "log", "event_description", "event_message",
"event_text", "fault_description", "fault_message", "fault_text",
"incident_description", "incident_message", "incident_text", "summary",
"описание", "комментарий", "сообщение", "детали", "деталь",
"примечание", "примечания",
"opis", "komentar", "poruka", "napomena",
}
_TYPE_PREFIX = {"service": "SVC", "host": "NODE", "person": "PERS"}
_IP_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
_EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}")
_HOST_RE = re.compile(r"\b(?:[a-zA-Z0-9\-]+\.){2,}[a-zA-Z]{2,}\b")
_HOST_HYPHEN_RE = re.compile(r"\b[a-zA-Z][a-zA-Z0-9_]{0,15}(?:-[a-zA-Z0-9_]+){1,4}-\d{1,4}\b")
def _normalize_col(name: str) -> str:
return re.sub(r"[^\w]+", "", str(name).lower(), flags=re.UNICODE)
def _col_tokens(name: str) -> set[str]:
# Split on anything that is not a letter or digit — crucially including
# underscore, unlike \w, so "start_time" tokenizes to {"start", "time"}
# rather than surviving as one opaque token that dodges the exclusion.
return set(re.split(r"[^0-9a-zA-Zа-яёА-ЯЁ]+", str(name).lower()))
def classify_column(name: str) -> Optional[str]:
if _col_tokens(name) & _EXCLUDE_TOKENS:
return None
norm = _normalize_col(name)
if any(kw in norm for kw in _SERVICE_HINTS):
return "service"
if any(kw in norm for kw in _HOST_HINTS):
return "host"
if any(kw in norm for kw in _PERSON_HINTS):
return "person"
return None
_NORM_DESCRIPTION_HINTS = {_normalize_col(h) for h in _DESCRIPTION_HINTS}
def is_description_column(name: str) -> bool:
return _normalize_col(name) in _NORM_DESCRIPTION_HINTS
def mask_tokens(text: str) -> str:
"""Regex-only pass for IP / hostname / email inside free text — the same
patterns the OpsLab parser applies server-side (belt and suspenders)."""
text = _EMAIL_RE.sub("[EMAIL]", text)
text = _IP_RE.sub("[IP]", text)
text = _HOST_RE.sub("[HOST]", text)
text = _HOST_HYPHEN_RE.sub("[HOST]", text)
return text
# ---------------------------------------------------------------------------
# File readers — CSV/TSV/TXT/JSON/XLSX in, header + list[dict] rows out.
# ---------------------------------------------------------------------------
def read_csv_like(path: str) -> tuple[list[str], list[dict]]:
with open(path, "rb") as f:
raw = f.read()
text = raw.decode("utf-8-sig", errors="replace")
first_line = text.split("\n", 1)[0]
delimiter = "\t" if first_line.count("\t") > first_line.count(",") else ","
reader = csv.DictReader(io.StringIO(text), delimiter=delimiter)
header = reader.fieldnames or []
rows = [dict(r) for r in reader]
return list(header), rows
def read_json(path: str) -> tuple[list[str], list[dict]]:
with open(path, "rb") as f:
data = json.load(f)
if isinstance(data, dict):
for key in ("data", "rows", "records", "items"):
if isinstance(data.get(key), list):
data = data[key]
break
if not isinstance(data, list) or not data:
return [], []
header = list(data[0].keys())
rows = [{k: ("" if v is None else v) for k, v in row.items()} for row in data]
return header, rows
_CELL_REF_RE = re.compile(r"([A-Z]+)(\d+)")
def _col_to_index(ref: str) -> int:
"""'C7' -> 2 (0-based column index from an Excel cell reference)."""
m = _CELL_REF_RE.match(ref)
letters = m.group(1) if m else ref
idx = 0
for ch in letters:
idx = idx * 26 + (ord(ch) - ord("A") + 1)
return idx - 1
def _parse_xml_part(raw: bytes) -> ET.Element:
"""ET.fromstring around Python's stdlib expat is not memory-safe against a
crafted DOCTYPE (XXE / billion-laughs). A genuine xlsx part written by
Excel or any spreadsheet library never has a DOCTYPE, so refusing one
is a free, dependency-free guard (defusedxml is not an option here —
the whole point of this script is zero pip installs on a locked-down
client machine)."""
if b" tuple[list[str], list[dict]]:
"""Minimal XLSX reader: stdlib zipfile + xml only, first worksheet.
A real .xlsx file is a zip archive of XML parts, so this needs no
external library (openpyxl is not something a client machine under a
strict IT-security policy can be expected to pip-install).
"""
ns_uri = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
ns = {"m": ns_uri}
with zipfile.ZipFile(path) as z:
shared: list[str] = []
if "xl/sharedStrings.xml" in z.namelist():
root = _parse_xml_part(z.read("xl/sharedStrings.xml"))
for si in root.findall("m:si", ns):
text = "".join(t.text or "" for t in si.iter("{%s}t" % ns_uri))
shared.append(text)
sheet_names = sorted(
n for n in z.namelist()
if n.startswith("xl/worksheets/sheet") and n.endswith(".xml")
)
if not sheet_names:
raise ValueError("No worksheet found in .xlsx")
root = _parse_xml_part(z.read(sheet_names[0]))
grid: dict[int, dict[int, str]] = {}
max_col = 0
for row_el in root.iter("{%s}row" % ns_uri):
r = int(row_el.get("r"))
for cell in row_el.findall("m:c", ns):
ref = cell.get("r", "")
col_idx = _col_to_index(ref) if ref else 0
max_col = max(max_col, col_idx)
cell_type = cell.get("t")
v_el = cell.find("m:v", ns)
if cell_type == "s" and v_el is not None:
value = shared[int(v_el.text)]
elif cell_type == "inlineStr":
is_el = cell.find("m:is", ns)
value = "".join(t.text or "" for t in is_el.iter("{%s}t" % ns_uri)) if is_el is not None else ""
elif v_el is not None:
value = v_el.text or ""
else:
value = ""
grid.setdefault(r, {})[col_idx] = value
if not grid:
return [], []
row_nums = sorted(grid.keys())
header_row = grid[row_nums[0]]
header = [header_row.get(c, f"col{c}") for c in range(max_col + 1)]
rows = []
for r in row_nums[1:]:
row = grid[r]
rows.append({header[c]: row.get(c, "") for c in range(max_col + 1)})
return header, rows
def read_input(path: str) -> tuple[list[str], list[dict]]:
ext = os.path.splitext(path)[1].lower()
if ext == ".xlsx":
return read_xlsx(path)
if ext == ".json":
return read_json(path)
return read_csv_like(path) # .csv, .txt, and anything else: CSV/TSV sniffing
# ---------------------------------------------------------------------------
# Dictionary persistence — salt + (type, original) -> pseudonym map, so a
# repeat run on the same client produces the same pseudonyms and two audits
# stay comparable.
# ---------------------------------------------------------------------------
def load_dict(path: str) -> tuple[str, dict[tuple[str, str], str]]:
salt = ""
mapping: dict[tuple[str, str], str] = {}
if not path or not os.path.exists(path):
return salt, mapping
with open(path, "r", encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
for row in reader:
if row.get("type") == "_salt":
salt = row.get("original", "")
continue
mapping[(row["type"], row["original"])] = row["pseudonym"]
return salt, mapping
def save_dict(path: str, salt: str, mapping: dict[tuple[str, str], str]) -> None:
with open(path, "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(["type", "original", "pseudonym"])
writer.writerow(["_salt", salt, ""])
for (col_type, original), pseudonym in sorted(mapping.items(), key=lambda kv: kv[1]):
writer.writerow([col_type, original, pseudonym])
def pseudonym_for(value: str, col_type: str, salt: str, mapping: dict[tuple[str, str], str]) -> str:
key = (col_type, value)
if key in mapping:
return mapping[key]
digest = hashlib.sha256(f"{salt}:{col_type}:{value}".encode("utf-8")).hexdigest()[:6].upper()
pseudonym = f"{_TYPE_PREFIX[col_type]}-{digest}"
while pseudonym in mapping.values(): # astronomically unlikely, cheap to guard
digest = hashlib.sha256(f"{salt}:{digest}".encode("utf-8")).hexdigest()[:6].upper()
pseudonym = f"{_TYPE_PREFIX[col_type]}-{digest}"
mapping[key] = pseudonym
return pseudonym
# ---------------------------------------------------------------------------
# Main pass
# ---------------------------------------------------------------------------
def anonymize(header: list[str], rows: list[dict], salt: str, mapping: dict[tuple[str, str], str]):
col_types = {col: classify_column(col) for col in header}
desc_cols = [col for col in header if is_description_column(col)]
cat_cols = {col: t for col, t in col_types.items() if t is not None}
# Pass 1: pseudonymize categorical columns in place.
for row in rows:
for col, col_type in cat_cols.items():
raw = row.get(col, "")
if raw is None or str(raw).strip() == "":
continue
value = str(raw).strip()
row[col] = pseudonym_for(value, col_type, salt, mapping)
# Pass 2: free-text columns — regex-mask IP/host/email, then replace any
# known categorical value that shows up inside the text. Longest names
# first so "ККС 015 Переводы" doesn't get shadowed by a shorter "ККС".
known_values = sorted(
((orig, pseudo) for (_, orig), pseudo in mapping.items() if len(orig) >= 3),
key=lambda pair: len(pair[0]),
reverse=True,
)
for row in rows:
for col in desc_cols:
raw = row.get(col, "")
if raw is None:
continue
text = str(raw)
if not text:
continue
text = mask_tokens(text)
for orig, pseudo in known_values:
if orig in text:
text = text.replace(orig, pseudo)
row[col] = text
return header, rows, cat_cols
_UPLOAD_URL = "https://opslab.consulting/#run"
def process_file(input_path: str, out_path: str, dict_path: str) -> dict:
"""Run the full anonymize pass and write both output files.
Shared by the CLI and the GUI so the two entry points can never drift —
each is a thin wrapper that collects a path and calls this.
Returns a small summary dict; raises on unreadable input so each
caller can report the error its own way (stderr vs. a message box).
"""
salt, mapping = load_dict(dict_path)
if not salt:
salt = secrets.token_hex(16)
header, rows = read_input(input_path)
if not header:
raise ValueError(f"Could not read any rows from {input_path}")
header, rows, cat_cols = anonymize(header, rows, salt, mapping)
with open(out_path, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=header)
writer.writeheader()
for row in rows:
writer.writerow(row)
save_dict(dict_path, salt, mapping)
service_pseudonyms = sorted(
pseudo for (col_type, _), pseudo in mapping.items() if col_type == "service"
)
return {
"out_path": out_path,
"dict_path": dict_path,
"cat_cols": cat_cols,
"service_pseudonyms": service_pseudonyms,
}
def _print_summary(result: dict) -> None:
print(f"Anonymized file written to: {result['out_path']}")
print(f"Dictionary (KEEP THIS YOURSELF, do not send it to OpsLab): {result['dict_path']}")
if not result["cat_cols"]:
print("\nNote: no service/host/people column was recognized by name in this "
"file — nothing was pseudonymized outside free-text IP/host/email masking. "
"Check the column headers if you expected names to be replaced.")
if result["service_pseudonyms"]:
print("\nCritical services field on the upload form — use these pseudonyms:")
print(", ".join(result["service_pseudonyms"]))
def _cli_main() -> None:
parser = argparse.ArgumentParser(
description="Replace service/host/people names with stable pseudonyms before sending a log to OpsLab.",
)
parser.add_argument("input", help="incident log file: .csv, .txt, .json or .xlsx")
parser.add_argument("--dict", dest="dict_path", default=None,
help="dictionary CSV path (default: .dict.csv). "
"Reusing it on a later run keeps the same pseudonyms.")
parser.add_argument("-o", "--output", dest="output", default=None,
help="output CSV path (default: .anon.csv)")
args = parser.parse_args()
base, _ = os.path.splitext(args.input)
out_path = args.output or f"{base}.anon.csv"
dict_path = args.dict_path or f"{base}.dict.csv"
try:
result = process_file(args.input, out_path, dict_path)
except ValueError as exc:
print(str(exc), file=sys.stderr)
sys.exit(1)
_print_summary(result)
def _gui_main() -> None:
"""Double-click entry point: no terminal, so every step is a dialog.
Picking the file and picking where to save it are both explicit user
choices (file pickers), not silent defaults — that matters here more
than in a typical script, because the whole point is that the client
stays in control of what happens to their data. The dictionary file
(the one that must NOT be sent to OpsLab) is saved automatically next
to the anonymized output, named clearly, and called out by name in the
final summary so it's never confused with the file to upload.
"""
import tkinter as tk
from tkinter import filedialog, messagebox
root = tk.Tk()
root.withdraw()
try:
input_path = filedialog.askopenfilename(
title="OpsLab anonymizer — choose your incident log file",
filetypes=[
("Incident log", "*.csv *.txt *.json *.xlsx"),
("All files", "*.*"),
],
)
if not input_path:
return # cancelled
base = os.path.splitext(os.path.basename(input_path))[0]
suggested_out = f"{base}.anon.csv"
out_path = filedialog.asksaveasfilename(
title="OpsLab anonymizer — save the anonymized file",
initialdir=os.path.dirname(input_path),
initialfile=suggested_out,
defaultextension=".csv",
filetypes=[("CSV", "*.csv")],
)
if not out_path:
return # cancelled
dict_path = os.path.splitext(out_path)[0] + ".dict.csv"
result = process_file(input_path, out_path, dict_path)
summary = (
"Done.\n\n"
f"Anonymized file (send this one to OpsLab):\n{result['out_path']}\n\n"
f"Dictionary — keep this yourself, do NOT send it:\n{result['dict_path']}"
)
if result["service_pseudonyms"]:
summary += "\n\nCritical services field on the upload form:\n" + \
", ".join(result["service_pseudonyms"])
if not result["cat_cols"]:
summary += ("\n\nNote: no service/host/people column was recognized by name — "
"check the column headers if you expected names to be replaced.")
if messagebox.askyesno("OpsLab anonymizer", summary + "\n\nOpen the OpsLab upload page now?"):
import webbrowser
webbrowser.open(_UPLOAD_URL)
except ValueError as exc:
messagebox.showerror("OpsLab anonymizer", str(exc))
except Exception as exc: # noqa: BLE001 — last resort so double-click never just vanishes
messagebox.showerror("OpsLab anonymizer", f"Something went wrong:\n{exc}")
finally:
root.destroy()
def main() -> None:
# No arguments means someone double-clicked the file rather than running
# it from a terminal — switch to a dialog-driven flow. `-h`/`--help` and
# any real argument still go through the normal CLI.
if len(sys.argv) == 1:
try:
_gui_main()
except ImportError:
print("tkinter is not available on this Python install.\n"
"Run from a terminal instead:\n"
" python opslab-anonymize.py ", file=sys.stderr)
sys.exit(1)
else:
_cli_main()
if __name__ == "__main__":
main()