wehub-resource-sync
957 lines
34 KiB
Python
957 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a security ownership map from git history."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import datetime as dt
|
|
import fnmatch
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
DEFAULT_SENSITIVE_RULES: list[tuple[str, str, float]] = [
|
|
("**/auth/**", "auth", 1.0),
|
|
("**/oauth/**", "auth", 1.0),
|
|
("**/rbac/**", "auth", 1.0),
|
|
("**/session/**", "auth", 1.0),
|
|
("**/token/**", "auth", 1.0),
|
|
("**/crypto/**", "crypto", 1.0),
|
|
("**/tls/**", "crypto", 1.0),
|
|
("**/ssl/**", "crypto", 1.0),
|
|
("**/secrets/**", "secrets", 1.0),
|
|
("**/keys/**", "secrets", 1.0),
|
|
("**/*.pem", "secrets", 1.0),
|
|
("**/*.key", "secrets", 1.0),
|
|
("**/*.p12", "secrets", 1.0),
|
|
("**/*.pfx", "secrets", 1.0),
|
|
("**/iam/**", "auth", 1.0),
|
|
("**/sso/**", "auth", 1.0),
|
|
]
|
|
|
|
DEFAULT_AUTHOR_EXCLUDE_REGEXES = [
|
|
"dependabot",
|
|
]
|
|
|
|
DEFAULT_COCHANGE_EXCLUDES = [
|
|
"**/Cargo.lock",
|
|
"**/Cargo.toml",
|
|
"**/package-lock.json",
|
|
"**/yarn.lock",
|
|
"**/pnpm-lock.yaml",
|
|
"**/go.sum",
|
|
"**/go.mod",
|
|
"**/Gemfile.lock",
|
|
"**/Pipfile.lock",
|
|
"**/poetry.lock",
|
|
"**/composer.lock",
|
|
"**/.github/**",
|
|
"**/.gitignore",
|
|
"**/.gitattributes",
|
|
"**/.gitmodules",
|
|
"**/.editorconfig",
|
|
"**/.vscode/**",
|
|
"**/.idea/**",
|
|
]
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Build ownership graphs and security ownership summaries from git history."
|
|
)
|
|
parser.add_argument("--repo", default=".", help="Path to the git repo (default: .)")
|
|
parser.add_argument(
|
|
"--out",
|
|
default="ownership-map-out",
|
|
help="Output directory for graph artifacts",
|
|
)
|
|
parser.add_argument("--since", default=None, help="Limit git log to commits since date")
|
|
parser.add_argument("--until", default=None, help="Limit git log to commits until date")
|
|
parser.add_argument(
|
|
"--identity",
|
|
choices=("author", "committer"),
|
|
default="author",
|
|
help="Identity to attribute touches to",
|
|
)
|
|
parser.add_argument(
|
|
"--date-field",
|
|
choices=("author", "committer"),
|
|
default="author",
|
|
help="Date field to use for recency and bucketing",
|
|
)
|
|
parser.add_argument(
|
|
"--include-merges",
|
|
action="store_true",
|
|
help="Include merge commits (excluded by default)",
|
|
)
|
|
parser.add_argument(
|
|
"--half-life-days",
|
|
type=float,
|
|
default=180.0,
|
|
help="Half life for recency weighting",
|
|
)
|
|
parser.add_argument(
|
|
"--sensitive-config",
|
|
default=None,
|
|
help="CSV file with pattern,tag,weight for sensitive paths",
|
|
)
|
|
parser.add_argument(
|
|
"--owner-threshold",
|
|
type=float,
|
|
default=0.5,
|
|
help="Share threshold for hidden owner detection",
|
|
)
|
|
parser.add_argument(
|
|
"--bus-factor-threshold",
|
|
type=int,
|
|
default=1,
|
|
help="Bus factor threshold for hotspots",
|
|
)
|
|
parser.add_argument(
|
|
"--stale-days",
|
|
type=int,
|
|
default=365,
|
|
help="Days since last touch to consider stale",
|
|
)
|
|
parser.add_argument(
|
|
"--min-touches",
|
|
type=int,
|
|
default=1,
|
|
help="Minimum touches to keep an edge",
|
|
)
|
|
parser.add_argument(
|
|
"--emit-commits",
|
|
action="store_true",
|
|
help="Write commit list to commits.jsonl",
|
|
)
|
|
parser.add_argument(
|
|
"--author-exclude-regex",
|
|
action="append",
|
|
default=[],
|
|
help="Regex for author name/email to exclude (repeatable)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-default-author-excludes",
|
|
action="store_true",
|
|
help="Disable default author excludes (dependabot)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-cochange",
|
|
action="store_true",
|
|
help="Disable co-change graph output",
|
|
)
|
|
parser.add_argument(
|
|
"--cochange-max-files",
|
|
type=int,
|
|
default=50,
|
|
help="Ignore commits touching more than this many files for co-change graph",
|
|
)
|
|
parser.add_argument(
|
|
"--cochange-min-count",
|
|
type=int,
|
|
default=2,
|
|
help="Minimum co-change count to keep file-file edge",
|
|
)
|
|
parser.add_argument(
|
|
"--cochange-min-jaccard",
|
|
type=float,
|
|
default=0.05,
|
|
help="Minimum Jaccard similarity to keep file-file edge",
|
|
)
|
|
parser.add_argument(
|
|
"--cochange-exclude",
|
|
action="append",
|
|
default=[],
|
|
help="Glob to exclude from co-change graph (repeatable)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-default-cochange-excludes",
|
|
action="store_true",
|
|
help="Disable default co-change excludes (lockfiles, .github, editor config)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-communities",
|
|
dest="communities",
|
|
action="store_false",
|
|
help="Disable community detection (enabled by default, requires networkx)",
|
|
)
|
|
parser.add_argument(
|
|
"--graphml",
|
|
action="store_true",
|
|
help="Emit ownership.graphml (requires networkx)",
|
|
)
|
|
parser.add_argument(
|
|
"--max-community-files",
|
|
type=int,
|
|
default=50,
|
|
help="Max files listed per community",
|
|
)
|
|
parser.add_argument(
|
|
"--community-top-owners",
|
|
type=int,
|
|
default=5,
|
|
help="Top maintainers saved per community",
|
|
)
|
|
parser.set_defaults(communities=True)
|
|
return parser.parse_args()
|
|
|
|
|
|
def load_sensitive_rules(path: str | None) -> list[tuple[str, str, float]]:
|
|
if not path:
|
|
return list(DEFAULT_SENSITIVE_RULES)
|
|
rules: list[tuple[str, str, float]] = []
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
for raw in handle:
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
parts = [part.strip() for part in line.split(",")]
|
|
if not parts:
|
|
continue
|
|
pattern = parts[0]
|
|
tag = parts[1] if len(parts) > 1 and parts[1] else "sensitive"
|
|
weight = float(parts[2]) if len(parts) > 2 and parts[2] else 1.0
|
|
rules.append((pattern, tag, weight))
|
|
return rules
|
|
|
|
|
|
def parse_date(value: str) -> dt.datetime:
|
|
parsed = dt.datetime.fromisoformat(value)
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=dt.timezone.utc)
|
|
return parsed
|
|
|
|
|
|
def offset_minutes(timestamp: dt.datetime) -> int | None:
|
|
offset = timestamp.utcoffset()
|
|
if offset is None:
|
|
return None
|
|
return int(offset.total_seconds() / 60)
|
|
|
|
|
|
def format_offset(minutes: int) -> str:
|
|
sign = "+" if minutes >= 0 else "-"
|
|
minutes = abs(minutes)
|
|
return f"{sign}{minutes // 60:02d}:{minutes % 60:02d}"
|
|
|
|
|
|
def recency_weighted(now: dt.datetime, when: dt.datetime, half_life_days: float) -> float:
|
|
if half_life_days <= 0:
|
|
return 1.0
|
|
age_days = max(0.0, (now - when).total_seconds() / 86400.0)
|
|
return math.exp(-math.log(2) * age_days / half_life_days)
|
|
|
|
|
|
def match_sensitive(path: str, rules: Iterable[tuple[str, str, float]]) -> dict[str, float]:
|
|
tags: dict[str, float] = defaultdict(float)
|
|
posix = path.replace("\\", "/")
|
|
for pattern, tag, weight in rules:
|
|
patterns = [pattern]
|
|
if pattern.startswith("**/"):
|
|
patterns.append(pattern[3:])
|
|
for candidate in patterns:
|
|
if fnmatch.fnmatchcase(posix, candidate):
|
|
tags[tag] += weight
|
|
break
|
|
return tags
|
|
|
|
|
|
def matches_glob(path: str, pattern: str) -> bool:
|
|
posix = path.replace("\\", "/")
|
|
patterns = [pattern]
|
|
if pattern.startswith("**/"):
|
|
patterns.append(pattern[3:])
|
|
return any(fnmatch.fnmatchcase(posix, candidate) for candidate in patterns)
|
|
|
|
|
|
def is_excluded(path: str, patterns: Iterable[str]) -> bool:
|
|
return any(matches_glob(path, pattern) for pattern in patterns)
|
|
|
|
|
|
def author_excluded(name: str, email: str, patterns: Iterable[re.Pattern[str]]) -> bool:
|
|
if not patterns:
|
|
return False
|
|
haystack = f"{name} {email}".strip()
|
|
return any(pattern.search(haystack) for pattern in patterns)
|
|
|
|
|
|
def compute_community_owners(
|
|
community_files: Iterable[str],
|
|
people: dict[str, dict[str, object]],
|
|
file_people_touches: dict[str, dict[str, int]],
|
|
file_people_recency: dict[str, dict[str, float]],
|
|
file_people_sensitive: dict[str, dict[str, float]],
|
|
top_n: int,
|
|
) -> dict[str, object]:
|
|
touches_by_person: dict[str, int] = defaultdict(int)
|
|
recency_by_person: dict[str, float] = defaultdict(float)
|
|
sensitive_by_person: dict[str, float] = defaultdict(float)
|
|
|
|
for path in community_files:
|
|
for person, touches in file_people_touches.get(path, {}).items():
|
|
touches_by_person[person] += touches
|
|
for person, recency in file_people_recency.get(path, {}).items():
|
|
recency_by_person[person] += recency
|
|
for person, weight in file_people_sensitive.get(path, {}).items():
|
|
sensitive_by_person[person] += weight
|
|
|
|
total_touches = sum(touches_by_person.values())
|
|
total_recency = sum(recency_by_person.values())
|
|
total_sensitive = sum(sensitive_by_person.values())
|
|
|
|
ranked = sorted(touches_by_person.items(), key=lambda item: item[1], reverse=True)
|
|
owners = []
|
|
for person_id, touches in ranked[:top_n]:
|
|
recency = recency_by_person.get(person_id, 0.0)
|
|
sensitive = sensitive_by_person.get(person_id, 0.0)
|
|
owners.append(
|
|
{
|
|
"person_id": person_id,
|
|
"name": people.get(person_id, {}).get("name", person_id),
|
|
"touches": touches,
|
|
"touch_share": round(touches / total_touches, 4) if total_touches else 0.0,
|
|
"recency_share": round(recency / total_recency, 4) if total_recency else 0.0,
|
|
"sensitive_share": round(sensitive / total_sensitive, 4)
|
|
if total_sensitive
|
|
else 0.0,
|
|
"primary_tz_offset": people.get(person_id, {}).get("primary_tz_offset", ""),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"bus_factor": len(touches_by_person),
|
|
"owner_count": len(touches_by_person),
|
|
"totals": {
|
|
"touches": total_touches,
|
|
"recency_weight": round(total_recency, 6),
|
|
"sensitive_weight": round(total_sensitive, 2),
|
|
},
|
|
"top_maintainers": owners,
|
|
}
|
|
|
|
|
|
def run_git_log(
|
|
repo: str, since: str | None, until: str | None, include_merges: bool
|
|
) -> Iterable[list[str]]:
|
|
cmd = [
|
|
"git",
|
|
"-C",
|
|
repo,
|
|
"log",
|
|
"--name-only",
|
|
"--no-renames",
|
|
"--date=iso-strict",
|
|
"--format=---%n%H%n%P%n%an%n%ae%n%ad%n%cn%n%ce%n%cd",
|
|
]
|
|
if not include_merges:
|
|
cmd.append("--no-merges")
|
|
if since:
|
|
cmd.extend(["--since", since])
|
|
if until:
|
|
cmd.extend(["--until", until])
|
|
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
assert proc.stdout is not None
|
|
|
|
batch: list[str] = []
|
|
for line in proc.stdout:
|
|
batch.append(line.rstrip("\n"))
|
|
if line.rstrip("\n") == "---" and len(batch) > 1:
|
|
yield batch[:-1]
|
|
batch = ["---"]
|
|
|
|
if batch:
|
|
yield batch
|
|
|
|
stderr = proc.stderr.read() if proc.stderr else ""
|
|
exit_code = proc.wait()
|
|
if exit_code != 0:
|
|
raise RuntimeError(stderr.strip() or "git log failed")
|
|
|
|
|
|
def iter_commits(lines: Iterable[list[str]]) -> Iterable[tuple[dict[str, object], list[str]]]:
|
|
for chunk in lines:
|
|
if not chunk or chunk[0] != "---":
|
|
continue
|
|
header = chunk[1:9]
|
|
if len(header) < 8:
|
|
continue
|
|
parents = [entry for entry in header[1].split(" ") if entry]
|
|
commit = {
|
|
"hash": header[0],
|
|
"parents": parents,
|
|
"is_merge": len(parents) > 1,
|
|
"author_name": header[2],
|
|
"author_email": header[3],
|
|
"author_date": header[4],
|
|
"committer_name": header[5],
|
|
"committer_email": header[6],
|
|
"committer_date": header[7],
|
|
}
|
|
files = [line for line in chunk[9:] if line.strip()]
|
|
yield commit, files
|
|
|
|
|
|
def ensure_out_dir(path: str) -> Path:
|
|
out_dir = Path(path)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
return out_dir
|
|
|
|
|
|
def write_csv(path: Path, header: list[str], rows: Iterable[list[str]]) -> None:
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.writer(handle)
|
|
writer.writerow(header)
|
|
for row in rows:
|
|
writer.writerow(row)
|
|
|
|
|
|
def build_ownership_map(args: argparse.Namespace) -> Path:
|
|
now = dt.datetime.now(dt.timezone.utc)
|
|
rules = load_sensitive_rules(args.sensitive_config)
|
|
out_dir = ensure_out_dir(args.out)
|
|
|
|
people: dict[str, dict[str, object]] = {}
|
|
files: dict[str, dict[str, object]] = {}
|
|
edges: dict[tuple[str, str], dict[str, object]] = {}
|
|
file_people_touches: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
|
file_people_recency: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
|
|
file_people_sensitive: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
|
|
tag_totals: dict[str, float] = defaultdict(float)
|
|
tag_person_totals: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
|
|
person_timezone_counts: dict[str, dict[int, int]] = defaultdict(lambda: defaultdict(int))
|
|
cochange_counts: dict[tuple[str, str], int] = defaultdict(int)
|
|
cochange_file_commits: dict[str, int] = defaultdict(int)
|
|
cochange_commits_used = 0
|
|
cochange_commits_skipped = 0
|
|
cochange_commits_filtered = 0
|
|
cochange_files_excluded = 0
|
|
|
|
commits_path = out_dir / "commits.jsonl"
|
|
commit_handle = None
|
|
if args.emit_commits:
|
|
commit_handle = commits_path.open("w", encoding="utf-8")
|
|
|
|
total_commits_seen = 0
|
|
total_commits_included = 0
|
|
commits_excluded_identities = 0
|
|
commits_excluded_merges = 0
|
|
total_edges = 0
|
|
|
|
author_exclude_regexes = []
|
|
if not args.no_default_author_excludes:
|
|
author_exclude_regexes.extend(DEFAULT_AUTHOR_EXCLUDE_REGEXES)
|
|
author_exclude_regexes.extend(args.author_exclude_regex)
|
|
author_exclude_patterns = [
|
|
re.compile(pattern, re.IGNORECASE) for pattern in author_exclude_regexes
|
|
]
|
|
|
|
cochange_excludes = []
|
|
if not args.no_default_cochange_excludes:
|
|
cochange_excludes.extend(DEFAULT_COCHANGE_EXCLUDES)
|
|
cochange_excludes.extend(args.cochange_exclude)
|
|
|
|
log_lines = run_git_log(args.repo, args.since, args.until, args.include_merges)
|
|
for commit, touched_files in iter_commits(log_lines):
|
|
total_commits_seen += 1
|
|
|
|
if commit.get("is_merge") and not args.include_merges:
|
|
commits_excluded_merges += 1
|
|
continue
|
|
|
|
identity_name = commit.get(f"{args.identity}_name", "")
|
|
identity_email = commit.get(f"{args.identity}_email", "")
|
|
if author_excluded(
|
|
identity_name,
|
|
identity_email,
|
|
author_exclude_patterns,
|
|
):
|
|
commits_excluded_identities += 1
|
|
continue
|
|
|
|
if not touched_files:
|
|
continue
|
|
|
|
total_commits_included += 1
|
|
if commit_handle:
|
|
commit_handle.write(json.dumps({**commit, "files": touched_files}) + "\n")
|
|
|
|
identity_name = commit.get(f"{args.identity}_name", "")
|
|
identity_email = commit.get(f"{args.identity}_email", "") or identity_name
|
|
commit_date = parse_date(commit.get(f"{args.date_field}_date", ""))
|
|
recency = recency_weighted(now, commit_date, args.half_life_days)
|
|
tz_minutes = offset_minutes(commit_date)
|
|
if tz_minutes is not None:
|
|
person_timezone_counts[identity_email][tz_minutes] += 1
|
|
unique_files = sorted(set(touched_files))
|
|
if not args.no_cochange and len(unique_files) > 1:
|
|
if len(unique_files) > args.cochange_max_files:
|
|
cochange_commits_skipped += 1
|
|
else:
|
|
filtered_files = [
|
|
path for path in unique_files if not is_excluded(path, cochange_excludes)
|
|
]
|
|
excluded = len(unique_files) - len(filtered_files)
|
|
if excluded:
|
|
cochange_files_excluded += excluded
|
|
if len(filtered_files) < 2:
|
|
cochange_commits_filtered += 1
|
|
if filtered_files:
|
|
for path in filtered_files:
|
|
cochange_file_commits[path] += 1
|
|
if len(filtered_files) >= 2:
|
|
cochange_commits_used += 1
|
|
for idx, path in enumerate(filtered_files):
|
|
for other in filtered_files[idx + 1 :]:
|
|
cochange_counts[(path, other)] += 1
|
|
|
|
person = people.setdefault(
|
|
identity_email,
|
|
{
|
|
"name": identity_name,
|
|
"email": identity_email,
|
|
"first_seen": commit_date,
|
|
"last_seen": commit_date,
|
|
"commit_count": 0,
|
|
"touches": 0,
|
|
"sensitive_touches": 0.0,
|
|
},
|
|
)
|
|
person["commit_count"] = int(person["commit_count"]) + 1
|
|
person["first_seen"] = min(person["first_seen"], commit_date)
|
|
person["last_seen"] = max(person["last_seen"], commit_date)
|
|
|
|
for path in touched_files:
|
|
file_entry = files.setdefault(
|
|
path,
|
|
{
|
|
"path": path,
|
|
"first_seen": commit_date,
|
|
"last_seen": commit_date,
|
|
"commit_count": 0,
|
|
"touches": 0,
|
|
"authors": set(),
|
|
"sensitive_tags": {},
|
|
},
|
|
)
|
|
file_entry["commit_count"] = int(file_entry["commit_count"]) + 1
|
|
file_entry["first_seen"] = min(file_entry["first_seen"], commit_date)
|
|
file_entry["last_seen"] = max(file_entry["last_seen"], commit_date)
|
|
file_entry["touches"] = int(file_entry["touches"]) + 1
|
|
file_entry["authors"].add(identity_email)
|
|
|
|
edge = edges.setdefault(
|
|
(identity_email, path),
|
|
{
|
|
"touches": 0,
|
|
"first_seen": commit_date,
|
|
"last_seen": commit_date,
|
|
"recency_weight": 0.0,
|
|
"sensitive_weight": 0.0,
|
|
},
|
|
)
|
|
edge["touches"] = int(edge["touches"]) + 1
|
|
edge["first_seen"] = min(edge["first_seen"], commit_date)
|
|
edge["last_seen"] = max(edge["last_seen"], commit_date)
|
|
edge["recency_weight"] = float(edge["recency_weight"]) + recency
|
|
|
|
tags = match_sensitive(path, rules)
|
|
if tags:
|
|
file_entry["sensitive_tags"] = tags
|
|
sensitive_weight = sum(tags.values())
|
|
edge["sensitive_weight"] = float(edge["sensitive_weight"]) + sensitive_weight
|
|
person["sensitive_touches"] = float(person["sensitive_touches"]) + sensitive_weight
|
|
file_people_sensitive[path][identity_email] += sensitive_weight
|
|
for tag, weight in tags.items():
|
|
tag_totals[tag] += weight
|
|
tag_person_totals[tag][identity_email] += weight
|
|
|
|
person["touches"] = int(person["touches"]) + 1
|
|
file_people_touches[path][identity_email] += 1
|
|
file_people_recency[path][identity_email] += recency
|
|
total_edges += 1
|
|
|
|
if commit_handle:
|
|
commit_handle.close()
|
|
|
|
people_rows = []
|
|
for email, person in sorted(people.items()):
|
|
tz_counts = person_timezone_counts.get(email, {})
|
|
primary_tz_offset = ""
|
|
primary_tz_minutes = ""
|
|
timezone_offsets = ""
|
|
if tz_counts:
|
|
primary_tz_minutes_value = max(tz_counts.items(), key=lambda item: (item[1], item[0]))[
|
|
0
|
|
]
|
|
primary_tz_offset = format_offset(primary_tz_minutes_value)
|
|
primary_tz_minutes = str(primary_tz_minutes_value)
|
|
timezone_offsets = ";".join(
|
|
f"{format_offset(minutes)}:{count}"
|
|
for minutes, count in sorted(tz_counts.items(), key=lambda item: item[0])
|
|
)
|
|
person["primary_tz_offset"] = primary_tz_offset
|
|
people_rows.append(
|
|
[
|
|
email,
|
|
str(person["name"]),
|
|
email,
|
|
person["first_seen"].isoformat(),
|
|
person["last_seen"].isoformat(),
|
|
str(person["commit_count"]),
|
|
str(person["touches"]),
|
|
f"{person['sensitive_touches']:.2f}",
|
|
primary_tz_offset,
|
|
primary_tz_minutes,
|
|
timezone_offsets,
|
|
]
|
|
)
|
|
|
|
file_rows = []
|
|
for path, file_entry in sorted(files.items()):
|
|
authors = file_entry["authors"]
|
|
bus_factor = len(authors)
|
|
tags = file_entry["sensitive_tags"]
|
|
tag_list = ";".join(sorted(tags.keys()))
|
|
sensitivity_score = sum(tags.values()) if tags else 0.0
|
|
file_rows.append(
|
|
[
|
|
path,
|
|
path,
|
|
file_entry["first_seen"].isoformat(),
|
|
file_entry["last_seen"].isoformat(),
|
|
str(file_entry["commit_count"]),
|
|
str(file_entry["touches"]),
|
|
str(bus_factor),
|
|
f"{sensitivity_score:.2f}",
|
|
tag_list,
|
|
]
|
|
)
|
|
|
|
edge_rows = []
|
|
for (email, path), edge in edges.items():
|
|
if int(edge["touches"]) < args.min_touches:
|
|
continue
|
|
edge_rows.append(
|
|
[
|
|
email,
|
|
path,
|
|
str(edge["touches"]),
|
|
f"{edge['recency_weight']:.6f}",
|
|
edge["first_seen"].isoformat(),
|
|
edge["last_seen"].isoformat(),
|
|
f"{edge['sensitive_weight']:.2f}",
|
|
]
|
|
)
|
|
|
|
cochange_rows: list[list[str]] = []
|
|
if not args.no_cochange:
|
|
for (file_a, file_b), count in cochange_counts.items():
|
|
if count < args.cochange_min_count:
|
|
continue
|
|
commits_a = cochange_file_commits.get(file_a, 0)
|
|
commits_b = cochange_file_commits.get(file_b, 0)
|
|
denom = commits_a + commits_b - count
|
|
if denom <= 0:
|
|
continue
|
|
jaccard = count / denom
|
|
if jaccard < args.cochange_min_jaccard:
|
|
continue
|
|
cochange_rows.append([file_a, file_b, str(count), f"{jaccard:.6f}"])
|
|
|
|
write_csv(
|
|
out_dir / "people.csv",
|
|
[
|
|
"person_id",
|
|
"name",
|
|
"email",
|
|
"first_seen",
|
|
"last_seen",
|
|
"commit_count",
|
|
"touches",
|
|
"sensitive_touches",
|
|
"primary_tz_offset",
|
|
"primary_tz_minutes",
|
|
"timezone_offsets",
|
|
],
|
|
people_rows,
|
|
)
|
|
write_csv(
|
|
out_dir / "files.csv",
|
|
[
|
|
"file_id",
|
|
"path",
|
|
"first_seen",
|
|
"last_seen",
|
|
"commit_count",
|
|
"touches",
|
|
"bus_factor",
|
|
"sensitivity_score",
|
|
"sensitivity_tags",
|
|
],
|
|
file_rows,
|
|
)
|
|
write_csv(
|
|
out_dir / "edges.csv",
|
|
[
|
|
"person_id",
|
|
"file_id",
|
|
"touches",
|
|
"recency_weight",
|
|
"first_seen",
|
|
"last_seen",
|
|
"sensitive_weight",
|
|
],
|
|
edge_rows,
|
|
)
|
|
if not args.no_cochange:
|
|
write_csv(
|
|
out_dir / "cochange_edges.csv",
|
|
[
|
|
"file_a",
|
|
"file_b",
|
|
"cochange_count",
|
|
"jaccard",
|
|
],
|
|
cochange_rows,
|
|
)
|
|
|
|
orphaned_sensitive_code = []
|
|
bus_factor_hotspots = []
|
|
for path, file_entry in files.items():
|
|
tags = file_entry["sensitive_tags"]
|
|
if not tags:
|
|
continue
|
|
bus_factor = len(file_entry["authors"])
|
|
last_seen = file_entry["last_seen"]
|
|
age_days = (now - last_seen).days
|
|
top_owner = None
|
|
if path in file_people_touches:
|
|
top_owner = max(file_people_touches[path].items(), key=lambda item: item[1])[0]
|
|
hotspot = {
|
|
"path": path,
|
|
"bus_factor": bus_factor,
|
|
"last_touch": last_seen.isoformat(),
|
|
"sensitivity_tags": sorted(tags.keys()),
|
|
"top_owner": top_owner,
|
|
}
|
|
if bus_factor <= args.bus_factor_threshold:
|
|
bus_factor_hotspots.append(hotspot)
|
|
if age_days >= args.stale_days:
|
|
orphaned_sensitive_code.append(
|
|
{
|
|
**hotspot,
|
|
"last_security_touch": last_seen.isoformat(),
|
|
}
|
|
)
|
|
|
|
hidden_owners = []
|
|
for tag, total in tag_totals.items():
|
|
if total <= 0:
|
|
continue
|
|
person_totals = tag_person_totals[tag]
|
|
if not person_totals:
|
|
continue
|
|
top_email, top_value = max(person_totals.items(), key=lambda item: item[1])
|
|
share = top_value / total
|
|
if share >= args.owner_threshold:
|
|
person_name = people.get(top_email, {}).get("name", top_email)
|
|
hidden_owners.append(
|
|
{
|
|
"person": top_email,
|
|
"name": person_name,
|
|
"controls": f"{share * 100:.0f}% of {tag} code",
|
|
"category": tag,
|
|
"share": round(share, 4),
|
|
}
|
|
)
|
|
|
|
summary = {
|
|
"generated_at": now.isoformat(),
|
|
"repo": os.path.abspath(args.repo),
|
|
"parameters": {
|
|
"since": args.since,
|
|
"until": args.until,
|
|
"half_life_days": args.half_life_days,
|
|
"bus_factor_threshold": args.bus_factor_threshold,
|
|
"stale_days": args.stale_days,
|
|
"owner_threshold": args.owner_threshold,
|
|
"sensitive_config": args.sensitive_config,
|
|
"identity": args.identity,
|
|
"date_field": args.date_field,
|
|
"include_merges": args.include_merges,
|
|
"cochange_enabled": not args.no_cochange,
|
|
"cochange_max_files": args.cochange_max_files,
|
|
"cochange_min_count": args.cochange_min_count,
|
|
"cochange_min_jaccard": args.cochange_min_jaccard,
|
|
"cochange_default_excludes": not args.no_default_cochange_excludes,
|
|
"cochange_excludes": cochange_excludes,
|
|
"author_default_excludes": not args.no_default_author_excludes,
|
|
"author_exclude_regexes": author_exclude_regexes,
|
|
"community_top_owners": args.community_top_owners,
|
|
},
|
|
"orphaned_sensitive_code": orphaned_sensitive_code,
|
|
"hidden_owners": hidden_owners,
|
|
"bus_factor_hotspots": bus_factor_hotspots,
|
|
"stats": {
|
|
"commits": total_commits_included,
|
|
"commits_seen": total_commits_seen,
|
|
"commits_excluded_identities": commits_excluded_identities,
|
|
"commits_excluded_merges": commits_excluded_merges,
|
|
"edges": total_edges,
|
|
"people": len(people),
|
|
"files": len(files),
|
|
"cochange_pairs_total": len(cochange_counts) if not args.no_cochange else 0,
|
|
"cochange_edges": len(cochange_rows) if not args.no_cochange else 0,
|
|
"cochange_commits_used": cochange_commits_used if not args.no_cochange else 0,
|
|
"cochange_commits_skipped": cochange_commits_skipped if not args.no_cochange else 0,
|
|
"cochange_commits_filtered": cochange_commits_filtered if not args.no_cochange else 0,
|
|
"cochange_files_excluded": cochange_files_excluded if not args.no_cochange else 0,
|
|
},
|
|
}
|
|
|
|
with (out_dir / "summary.json").open("w", encoding="utf-8") as handle:
|
|
json.dump(summary, handle, indent=2)
|
|
|
|
if args.communities or args.graphml:
|
|
try:
|
|
import networkx as nx
|
|
from networkx.algorithms import bipartite
|
|
except ImportError:
|
|
raise RuntimeError(
|
|
"networkx is required for communities/graphml output. Install with: pip install networkx"
|
|
)
|
|
else:
|
|
graph_bipartite = None
|
|
graph_cochange = None
|
|
person_nodes = set()
|
|
file_nodes = set()
|
|
community_index: dict[str, int] = {}
|
|
community_metadata: list[dict[str, object]] = []
|
|
|
|
if args.graphml or (args.communities and (args.no_cochange or not cochange_rows)):
|
|
graph_bipartite = nx.Graph()
|
|
for (email, path), edge in edges.items():
|
|
if int(edge["touches"]) < args.min_touches:
|
|
continue
|
|
graph_bipartite.add_node(email, node_type="person")
|
|
graph_bipartite.add_node(path, node_type="file")
|
|
graph_bipartite.add_edge(email, path, weight=float(edge["touches"]))
|
|
person_nodes.add(email)
|
|
file_nodes.add(path)
|
|
|
|
if not args.no_cochange and cochange_rows:
|
|
graph_cochange = nx.Graph()
|
|
for file_a, file_b, count, jaccard in cochange_rows:
|
|
graph_cochange.add_edge(
|
|
file_a,
|
|
file_b,
|
|
weight=float(jaccard),
|
|
count=int(count),
|
|
)
|
|
|
|
if args.communities:
|
|
communities_result = None
|
|
if graph_cochange is not None:
|
|
communities_result = list(
|
|
nx.algorithms.community.greedy_modularity_communities(
|
|
graph_cochange, weight="weight"
|
|
)
|
|
)
|
|
elif graph_bipartite is not None and file_nodes:
|
|
projected = bipartite.weighted_projected_graph(graph_bipartite, file_nodes)
|
|
communities_result = list(
|
|
nx.algorithms.community.greedy_modularity_communities(projected)
|
|
)
|
|
|
|
if communities_result is not None:
|
|
serialized = []
|
|
for idx, community in enumerate(communities_result, start=1):
|
|
files_list = sorted(community)
|
|
owners = compute_community_owners(
|
|
files_list,
|
|
people,
|
|
file_people_touches,
|
|
file_people_recency,
|
|
file_people_sensitive,
|
|
args.community_top_owners,
|
|
)
|
|
for path in files_list:
|
|
community_index[path] = idx
|
|
entry = {
|
|
"id": idx,
|
|
"size": len(files_list),
|
|
"files": files_list[: args.max_community_files],
|
|
"maintainers": owners["top_maintainers"],
|
|
"bus_factor": owners["bus_factor"],
|
|
"owner_count": owners["owner_count"],
|
|
"totals": owners["totals"],
|
|
}
|
|
serialized.append(entry)
|
|
metadata = dict(entry)
|
|
metadata.pop("files", None)
|
|
community_metadata.append(metadata)
|
|
with (out_dir / "communities.json").open("w", encoding="utf-8") as handle:
|
|
json.dump(serialized, handle, indent=2)
|
|
|
|
if args.communities:
|
|
for node, community_id in community_index.items():
|
|
if graph_cochange is not None and node in graph_cochange:
|
|
graph_cochange.nodes[node]["community_id"] = community_id
|
|
if graph_bipartite is not None and node in graph_bipartite:
|
|
graph_bipartite.nodes[node]["community_id"] = community_id
|
|
|
|
graph_for_json = graph_cochange or graph_bipartite
|
|
if graph_for_json is not None:
|
|
try:
|
|
from networkx.readwrite import json_graph
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
data = json_graph.node_link_data(graph_for_json, edges="edges")
|
|
data.setdefault("graph", {})
|
|
data["graph"]["community_maintainers"] = community_metadata
|
|
json_name = (
|
|
"cochange.graph.json"
|
|
if graph_for_json is graph_cochange
|
|
else "ownership.graph.json"
|
|
)
|
|
with (out_dir / json_name).open("w", encoding="utf-8") as handle:
|
|
json.dump(data, handle, indent=2)
|
|
|
|
if args.graphml:
|
|
if graph_bipartite is not None:
|
|
nx.write_graphml(graph_bipartite, out_dir / "ownership.graphml")
|
|
if graph_cochange is not None:
|
|
nx.write_graphml(graph_cochange, out_dir / "cochange.graphml")
|
|
|
|
return out_dir
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
try:
|
|
out_dir = build_ownership_map(args)
|
|
except RuntimeError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Ownership map written to {out_dir}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|