Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/compliance.md
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,9 @@ Two optional, keyword-only arguments add evidence when available:

- **`dataframe=`** — the source frame. Recovers per-column roles and missing
ratios via [`freshdata.infer_roles`](api-reference.md), sharpening the HIPAA and
ALCOA reports.
ALCOA reports. Without it, the full column list comes from the report's
`input_columns` (recorded by `freshdata.clean`). If neither is available, the
HIPAA report sets `coverage_verifiable` to `False`, adds a warning, and does not pass.
- **`enterprise_result=`** — an [enterprise](feature-overview.md) result supplying
the 0–100 Data Trust Score, PII-masking events, and fuzzy-clustering lineage.

Expand Down
3 changes: 3 additions & 0 deletions src/freshdata/cleaner.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,9 @@ def run_pipeline( # noqa: PLR0915 - fixed-order pipeline orchestration
if config.column_names:
out = normalize_column_names(out, report)
_emit_progress(progress_callback, "column_names", "after", out)
# Full column evidence for compliance reports, in the report's (post-rename)
# namespace, so untouched columns are still visible without the source frame.
report.input_columns = [str(c) for c in out.columns]

# Hard protected-column guard (context policy / mutable=False): fold the
# protected set into preserve_columns so drop/impute logic honors it, and
Expand Down
30 changes: 25 additions & 5 deletions src/freshdata/compliance/_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from __future__ import annotations

import logging
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import Any

import pandas as pd
Expand Down Expand Up @@ -158,6 +158,12 @@ class ComplianceContext:
core_action_count: int
clean_report: Any
input_dataframe_hash: str | None = None
#: ``True`` when ``all_columns`` is the full column list (from ``dataframe=``
#: or the report's recorded ``input_columns``); ``False`` when it was only
#: inferred from actions/masked columns, so untouched columns are invisible.
columns_complete: bool = False
#: ``{column: strategy}`` recorded by an enterprise mask report, when known.
mask_strategies: dict[str, str] = field(default_factory=dict)


# --------------------------------------------------------------------------- #
Expand Down Expand Up @@ -297,6 +303,9 @@ def build_context(

# Trust score: EnterpriseResult -> domain score (0–1, scaled) -> config.
trust_score = _resolve_trust_score(enterprise, clean_report, config)
all_columns, columns_complete = _resolve_all_columns(
clean_report, dataframe, core_actions, masked_columns
)

return ComplianceContext(
session_id=new_session_id(),
Expand All @@ -308,10 +317,12 @@ def build_context(
roles=roles,
missing_ratio=missing_ratio,
domain_sensitive_columns=domain_sensitive,
all_columns=_resolve_all_columns(clean_report, dataframe, core_actions, masked_columns),
all_columns=all_columns,
core_action_count=len(core_actions),
clean_report=clean_report,
input_dataframe_hash=config.input_dataframe_hash,
columns_complete=columns_complete,
mask_strategies={c: str(s) for c, s in mask_meta.items() if s},
)


Expand All @@ -335,14 +346,23 @@ def _resolve_all_columns(
dataframe: pd.DataFrame | None,
core_actions: list[Any],
masked_columns: set[str],
) -> list[str]:
) -> tuple[list[str], bool]:
"""Return ``(columns, complete)``.

``complete`` is ``True`` only when the full column list is known: from the
source ``dataframe`` or the report's recorded ``input_columns``. Otherwise
columns are inferred from what cleaning touched, which misses untouched ones.
"""
if dataframe is not None:
return [str(c) for c in dataframe.columns]
return [str(c) for c in dataframe.columns], True
recorded = getattr(clean_report, "input_columns", None) or []
if recorded:
return [str(c) for c in recorded], True
columns: set[str] = set(masked_columns)
for action in core_actions:
column = getattr(action, "column", None)
if column:
columns.add(column)
for attr in ("columns_dropped", "columns_imputed", "columns_preserved"):
columns.update(getattr(clean_report, attr, []) or [])
return sorted(columns)
return sorted(columns), False
26 changes: 20 additions & 6 deletions src/freshdata/compliance/_gdpr.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,25 @@
"drop_row": "Art.5(1)(e) storage limitation (empty/irreparable record)",
"pii_mask": "Art.5(1)(f) integrity and confidentiality",
}
_SECURITY_MEASURES = [
"Hash-salt PII masking (SHA-256 + salt)",
"Audit trail generation",
"Access-gated Data Trust Score",
]


def _security_measures(ctx: ComplianceContext) -> list[str]:
"""Security measures actually evidenced by this run (never a fixed list).

Masking is listed only when columns were masked, naming the strategies that
were recorded; no cryptographic properties are asserted.
"""
measures: list[str] = []
masked = set(ctx.masked_columns)
masked.update(a.column for a in ctx.actions if a.action_type == "pii_mask" and a.column)
if masked:
strategies = sorted({s for c, s in ctx.mask_strategies.items() if c in masked and s})
detail = f" (strategies: {', '.join(strategies)})" if strategies else ""
measures.append(f"PII masking/anonymisation applied to {len(masked)} column(s){detail}")
measures.append("Audit trail generation")
if ctx.trust_score_available:
measures.append("Data Trust Score computed")
return measures


def _personal_data_categories(ctx: ComplianceContext) -> list[str]:
Expand Down Expand Up @@ -62,7 +76,7 @@ def generate_gdpr(ctx: ComplianceContext, config: ComplianceConfig) -> Framework
"third_country_transfers": False,
"third_country_transfers_note": ("freshdata processes in-memory; no network transfer."),
"retention_days": config.retention_days,
"security_measures": list(_SECURITY_MEASURES),
"security_measures": _security_measures(ctx),
"automated_decision_making": True,
"automated_decision_making_note": ("Column-level cleaning decisions are fully automated."),
"safeguards": (
Expand Down
71 changes: 65 additions & 6 deletions src/freshdata/compliance/_hipaa.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from __future__ import annotations

import re
from typing import Any

from ._adapter import ComplianceContext
Expand All @@ -25,12 +26,19 @@
"names": {
"id": 1,
"description": "Names",
"detection_hints": ["name", "first_name", "last_name", "full_name", "patient_name"],
"detection_hints": [
"name",
"first_name",
"last_name",
"full_name",
"patient_name",
"surname",
],
},
"geographic": {
"id": 2,
"description": "Geographic subdivisions smaller than state",
"detection_hints": ["address", "street", "city", "county", "zip", "postal"],
"detection_hints": ["address", "street", "city", "county", "zip", "zipcode", "postal"],
},
"dates": {
"id": 3,
Expand Down Expand Up @@ -126,6 +134,46 @@
}


#: Hints this short (``ip``, ``sin``, ``date``, ...) are ambiguous inside other
#: words, so they only match as whole name tokens.
_SHORT_HINT_MAX_LEN = 4
_TOKEN_BOUNDARY = re.compile(
r"(?<=[a-z])(?=[A-Z])" # camelCase
r"|(?<=[A-Z])(?=[A-Z][a-z]{2})" # acronym followed by a word: IPAddress
r"|(?<=[A-Za-z])(?=[0-9])|(?<=[0-9])(?=[A-Za-z])" # letter/digit: zip5, ipv4
)
_SEPARATORS = re.compile(r"[\W_]+")


def _tokens(name: object) -> tuple[str, ...]:
"""Split a column name into lowercase tokens (separators, camelCase, digits)."""
spaced = _TOKEN_BOUNDARY.sub(" ", str(name))
return tuple(t.lower() for t in _SEPARATORS.split(spaced) if t)


def _hint_matches(hint_tokens: tuple[str, ...], col_tokens: tuple[str, ...]) -> bool:
"""Whether the hint (as ``hint_tokens``) names a column with ``col_tokens``.

Every hint matches as a whole token or token sequence (``date_of_birth``
matches ``DateOfBirth``). Longer hints may also appear inside run-together
names (``patientemail``) but never starting part-way through a token they
extend past (``ip_address`` does not match ``ship_address``).
"""
n = len(hint_tokens)
if any(col_tokens[i : i + n] == hint_tokens for i in range(len(col_tokens) - n + 1)):
return True
compact = "".join(hint_tokens)
if len(compact) <= _SHORT_HINT_MAX_LEN:
return False
joined = "".join(col_tokens)
offset = 0
for token in col_tokens:
if compact in token or joined.startswith(compact, offset):
return True
offset += len(token)
return False


def _known_columns(ctx: ComplianceContext) -> list[str]:
columns: set[str] = set(ctx.all_columns) | set(ctx.masked_columns)
columns.update(a.column for a in ctx.actions if a.column)
Expand All @@ -134,16 +182,19 @@ def _known_columns(ctx: ComplianceContext) -> list[str]:

def generate_hipaa(ctx: ComplianceContext, config: ComplianceConfig) -> FrameworkReport:
known_columns = _known_columns(ctx)
column_tokens = {col: _tokens(col) for col in known_columns}
masked = set(ctx.masked_columns)

identifier_coverage: dict[str, dict] = {}
addressed = detected_not_addressed = not_detected = 0
gaps: list[str] = []

for key, spec in HIPAA_IDENTIFIERS.items():
hints = spec["detection_hints"]
hint_tokens = [_tokens(hint) for hint in spec["detection_hints"]]
columns_found = [
col for col in known_columns if any(hint in col.lower() for hint in hints)
col
for col in known_columns
if any(_hint_matches(tokens, column_tokens[col]) for tokens in hint_tokens)
]
columns_masked = [col for col in columns_found if col in masked]

Expand Down Expand Up @@ -180,6 +231,13 @@ def generate_hipaa(ctx: ComplianceContext, config: ComplianceConfig) -> Framewor
raise ComplianceGapError(f"HIPAA Safe Harbor gaps detected: {gaps}")

errors = [f"Identifier {key!r} detected but not addressed (no PII masking)." for key in gaps]
warnings: list[str] = []
if not ctx.columns_complete:
warnings.append(
"Column coverage not verifiable without dataframe=: the clean report records "
"no input column list, so identifier columns that cleaning did not touch "
"cannot be detected. Pass the source frame as dataframe= to verify coverage."
)

data = {
"report_id": new_entry_id("HIPAA"),
Expand All @@ -188,13 +246,14 @@ def generate_hipaa(ctx: ComplianceContext, config: ComplianceConfig) -> Framewor
"identifier_coverage": identifier_coverage,
"summary": summary,
"gaps": gaps,
"coverage_verifiable": ctx.columns_complete,
"caveat": _HIPAA_CAVEAT,
}
return FrameworkReport(
framework_key=FRAMEWORK_KEY,
framework_name=FRAMEWORK_NAME,
passed=not gaps,
warnings=[],
passed=not gaps and ctx.columns_complete,
warnings=warnings,
errors=errors,
data=data,
)
1 change: 1 addition & 0 deletions src/freshdata/repairplan.py
Original file line number Diff line number Diff line change
Expand Up @@ -719,6 +719,7 @@ def execute_plan(
rows_before=len(df),
cols_before=df.shape[1],
missing_before=int(df.isna().sum().sum()),
input_columns=[str(c) for c in df.columns],
)
out = df.copy(deep=False)
guard_snapshot = snapshot_protected(
Expand Down
6 changes: 6 additions & 0 deletions src/freshdata/report.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,12 @@ class CleanReport(HtmlReprMixin):
columns_dropped: list[str] = field(default_factory=list)
columns_imputed: list[str] = field(default_factory=list)
columns_preserved: list[str] = field(default_factory=list)
#: Every column the pipeline received, in input order, under the names used
#: throughout this report (i.e. after column-name normalization). Empty when
#: the producer did not record it. Compliance generators use it as column
#: evidence when no source ``dataframe=`` is supplied. Not part of
#: :meth:`to_dict` (the stable audit payload is unchanged).
input_columns: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
recommendations: list[str] = field(default_factory=list)
#: Per-cell record of values that ``fix_dtypes`` coerced to missing because
Expand Down
30 changes: 30 additions & 0 deletions tests/test_compliance/test_gdpr.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,33 @@ def test_article_30_record_fields(make_report):
assert article_30["third_country_transfers"] is False
assert article_30["automated_decision_making"] is True
assert gdpr.data["caveat"]


# --- #287: security measures reflect evidence, not a constant list ------------


def test_security_measures_omit_masking_when_nothing_masked(sample_report, sample_df):
gdpr = _gdpr(sample_report, dataframe=sample_df)
measures = gdpr.data["article_30"]["security_measures"]
assert not any("mask" in m.lower() for m in measures)
assert not any("sha-256" in m.lower() or "salt" in m.lower() for m in measures)
assert "Audit trail generation" in measures
# No enterprise trust score was supplied, so none is claimed.
assert not any("trust score" in m.lower() for m in measures)


def test_security_measures_list_masking_from_config(make_report):
report = make_report({"step": "missing", "column": "age", "count": 0})
gdpr = _gdpr(report, config=ComplianceConfig(masked_columns=["email", "ssn"]))
measures = gdpr.data["article_30"]["security_measures"]
assert measures[0] == "PII masking/anonymisation applied to 2 column(s)"


def test_security_measures_name_recorded_strategies(sample_report, enterprise_stub):
result = enterprise_stub(sample_report, overall=90.0, masked=["email"])
gdpr = _gdpr(result)
measures = gdpr.data["article_30"]["security_measures"]
assert measures[0] == (
"PII masking/anonymisation applied to 1 column(s) (strategies: sha256+salt)"
)
assert "Data Trust Score computed" in measures
Loading
Loading