"""Minimize identifying metadata before UI, analysis and export. This is a disclosure-reduction boundary, not PHI detection or HIPAA certification. Legacy inputs and provider-returned metadata can enter memory before projection. Only deployment-reviewed exact allowlists can authorize otherwise omitted names. """ from datetime import datetime, timezone import ipaddress import json from pathlib import Path, PureWindowsPath, PurePosixPath import re from samsara.analyzers.dependencies import (STANDARD_SERVICES, STANDARD_PROCESS, CLIENT_PROCESS, FAMILIES, ROLE_FAMILIES, service_family, normal_name) POLICY_PATH = Path(__file__).resolve().parents[1] / 'assets' / 'privacy_policy.json' PROJECT_ROOT = Path(__file__).resolve().parents[2] OMITTED = 'Omitted by privacy policy' SECTIONS = {'system','interfaces','roles','services','network','tasks','cron','software','storage', 'shares','share_activity','certificates','directory','ad','privacy'} SCORES = {'complete','partial','failed','not_collected'} ERROR_CODES = {'access_denied','command_unavailable','timeout','overall_timeout','output_limit', 'powershell_failed','invalid_output','serialization_failed','collection_failed','cleanup_failed'} KNOWN_ROLES = set(ROLE_FAMILIES) | {'net-framework-core','net-framework-45-core','net-framework-45-aspnet', 'net-framework-features','net-framework-45-features','powershell','powershell-v2','powershell-ise', 'windows-defender','windows-defender-features','storage-services','file-services','fs-fileserver', 'wow64-support','rsat','telnet-client','windows-search'} KNOWN_SERVICES = STANDARD_SERVICES | {name for _,names,_,_ in FAMILIES.values() for name in names} KNOWN_PROCESSES = STANDARD_PROCESS | CLIENT_PROCESS | {name for _,_,_,names in FAMILIES.values() for name in names} PRODUCTS = ('Microsoft SQL Server','PostgreSQL','MariaDB','MySQL','nginx','Apache HTTP Server', 'Redis','Docker','Microsoft .NET','Microsoft Visual C++','Microsoft Edge','Google Chrome', 'Mozilla Firefox','Steam','Valheim Dedicated Server','Minecraft Server') TASK_STATES = {'Unknown','Disabled','Queued','Ready','Running','active/waiting','active/running','inactive/dead'} SERVICE_STATES = {'Running','Stopped','Paused','Start Pending','Stop Pending','Unknown','active/exited','inactive/dead','failed/failed'} STARTUP = {'Auto','Automatic','Manual','Disabled','enabled','disabled','static','indirect','masked','generated','alias','Unknown'} NETWORK_STATES = {'Listen','Established','Bound','Closed','TimeWait','CloseWait','SynSent','SynReceived','FinWait1','FinWait2','LastAck','Closing','DeleteTCB'} RETAIN_KEYS = {'hostname','software_names','service_names','process_names','task_names', 'storage_names_and_roots','private_peer_addresses'} SOURCES = {'Local Windows read-only collection','Local Linux read-only collection'} def load_privacy_policy(path=POLICY_PATH): try: with Path(path).open('rb') as source: raw = source.read(65537) if len(raw)>65536: raise ValueError() policy = json.loads(raw.decode('utf-8')) expected = {'version','report_directory','approved_hostnames','approved_share_roots', 'approved_service_names','approved_software_names','max_session_assessments','retain'} if not isinstance(policy,dict) or set(policy)!=expected or policy['version']!='1.0': raise ValueError() if type(policy['max_session_assessments']) is not int or not 1<=policy['max_session_assessments']<=50: raise ValueError() retain=policy['retain'] if not isinstance(retain,dict) or set(retain)!=RETAIN_KEYS or any(type(v) is not bool for v in retain.values()): raise ValueError() if not isinstance(policy['report_directory'],str) or not policy['report_directory']: raise ValueError() for field in expected-{'version','report_directory','max_session_assessments','retain'}: if not isinstance(policy[field],list) or len(policy[field])>200 or any(not isinstance(v,str) or not v or len(v)>256 or any(ord(c)<32 for c in v) for v in policy[field]): raise ValueError() for root in policy['approved_share_roots']: canonical_share_root(root) # Exact roots only; never a descendant allowlist. return policy except (OSError, UnicodeError, ValueError, TypeError, RecursionError): raise ValueError('Privacy policy is unavailable or invalid. Collection/export is blocked; contact the deployment administrator.') from None def canonical_share_root(value): if not isinstance(value,str) or not value or any(ord(c)<32 for c in value): raise ValueError('Invalid approved share root.') if value.startswith(('\\\\','//')) or '..' in value.replace('\\','/').split('/'): raise ValueError('Only explicit local share roots may be approved.') windows = PureWindowsPath(value) if windows.drive: if not re.fullmatch(r'[A-Za-z]:',windows.drive) or not windows.is_absolute(): raise ValueError('Invalid approved local root.') return str(windows).rstrip('\\').casefold() posix = PurePosixPath(value) if not posix.is_absolute(): raise ValueError('Approved share roots must be absolute.') return str(posix).rstrip('/') or '/' def safe_time(value): if not isinstance(value,str) or not re.fullmatch(r'[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,7})?(?:Z|[+-][0-9]{2}:[0-9]{2})',value): return None try: parsed=datetime.fromisoformat(value.replace('Z','+00:00')) return parsed.astimezone(timezone.utc).isoformat() except ValueError: return None def numeric_version(value): if not isinstance(value,str): return 'Not collected' match=re.match(r'^[0-9]{1,5}(?:\.[0-9]{1,5}){0,3}(?![0-9.])',value) return match[0] if match else 'Not collected' def safe_ip(value): if not isinstance(value,str): return '' try: return str(ipaddress.ip_address(value.split('%')[0].strip('[]'))) except ValueError: return '' REMOTE_SCOPES = {'loopback','host','unspecified','private','public'} def remote_scope(value, own, previous=None, keep_private=False): """Classify a peer address. Only loopback, this host's own and wildcard addresses stay numeric: other peers (clients, patients, workstations) are reduced to private/public.""" if not isinstance(value,str) or not value.strip(): return (previous if previous in ('private','public') else ''), '' try: ip=ipaddress.ip_address(value.split('%')[0].strip().strip('[]')) except ValueError: return '', '' if isinstance(ip,ipaddress.IPv6Address) and ip.ipv4_mapped: ip=ip.ipv4_mapped if ip.is_unspecified: return 'unspecified', str(ip) if ip.is_loopback: return 'loopback', str(ip) if ip in own: return 'host', str(ip) if ip.is_private or ip.is_link_local: return 'private', (str(ip) if keep_private else '') return 'public', '' # --- Retained names ------------------------------------------------------------------- # Infrastructure names (software, services, shares, mount roots) are what an assessment # needs. A pattern filter withholds values that look like personal identifiers. This is a # heuristic, not PHI detection: administrators can switch each category off in `retain`. LABEL = re.compile(r"[A-Za-z0-9][A-Za-z0-9 ._()+#$:,&'@\-]{0,119}") UNIT_TEMPLATE = re.compile(r'[A-Za-z0-9_.-]+@[A-Za-z0-9_:\-]*(?:\.(?:service|timer|socket|path|mount))?') IDENTIFIER_PATTERNS = (re.compile(r'[0-9]{6,}'), re.compile(r'\b[0-9]{3}-[0-9]{2}-[0-9]{4}\b'), re.compile(r'\b[0-9]{3}[-. ][0-9]{3}[-. ][0-9]{4}\b')) def safe_label(value): if not isinstance(value,str): return None text=value.strip() if not LABEL.fullmatch(text) or ('@' in text and not UNIT_TEMPLATE.fullmatch(text)): return None if any(pattern.search(text) for pattern in IDENTIFIER_PATTERNS): return None return text def safe_path(value): """Share/mount roots only: every component must pass safe_label; no traversal; depth <= 4 components.""" if not isinstance(value,str) or not value or len(value)>260: return None norm=value.strip().replace('\\','/') drive=re.fullmatch(r'[A-Za-z]:(/.*)?',norm) if drive: parts=norm[2:].split('/') elif norm.startswith('//'): parts=norm[2:].split('/') elif re.fullmatch(r'[A-Za-z0-9._-]+:/.*',norm): parts=[norm.split(':',1)[0]]+norm.split(':',1)[1].split('/') elif norm.startswith('/'): parts=norm.split('/') else: return None parts=[part for part in parts if part] if len(parts)>4 or any(part in ('.','..') or safe_label(part) is None for part in parts): return None return value.strip() def hostname_label(value): text=safe_label(value) return text if text and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9.-]{0,252}',text) else None def integer(value, maximum=2**32-1): return value if type(value) is int and 0<=value<=maximum else 0 def service_label(value, index, policy): if not isinstance(value,str): return f'Service-{index+1:04d}' if value in policy['approved_service_names']: return value if policy['retain']['service_names'] and safe_label(value): return safe_label(value) name=normal_name(value) if name in KNOWN_SERVICES: return name family=service_family(value) if family: # Preserve family recognition without retaining instance/customer suffixes. representative=FAMILIES[family][1][0] return representative if re.fullmatch(r'(?:vmic|systemd-)[a-z0-9_.-]+',name): return f'OS service-{index+1:04d}' return f'Service-{index+1:04d}' def task_is_maintenance(row): path=str(row.get('path','')).replace('/','\\').lower() return path.startswith('\\microsoft\\windows\\') and not path.startswith('\\microsoft\\windows\\powershell\\scheduledjobs') def enum_value(value, choices, default='Unknown'): return value if isinstance(value,str) and value in choices else default def minimize_inventory(data, policy=None, *, trusted_source=None): policy=load_privacy_policy() if policy is None else policy if not isinstance(data,dict) or not isinstance(data.get('server'),dict) or not isinstance(data.get('inventory'),dict) or not isinstance(data.get('checks'),dict): raise ValueError('Invalid inventory container.') stamp=safe_time(data.get('collected_at')) if not stamp: raise ValueError('Invalid collection timestamp.') raw=data['server']; platform=raw.get('platform') if platform not in ('windows','linux'): raise ValueError('Unsupported inventory platform.') year=raw.get('os_version') if raw.get('os_version') in ('2016','2019','2022','2025') else None if platform=='windows' and year is None: match=re.search(r'\b(2016|2019|2022|2025)\b',str(raw.get('operating_system',''))) year=match[0] if match else None linux_name=next((label for pattern,label in ((r'\bred hat\b|\brhel\b','Red Hat Enterprise Linux'),(r'\brocky\b','Rocky Linux'),(r'\balmalinux\b','AlmaLinux'),(r'\boracle linux\b','Oracle Linux'),(r'\bcentos\b','CentOS'),(r'\blinux mint\b','Linux Mint'),(r'\bubuntu\b','Ubuntu'),(r'\bdebian\b','Debian')) if re.search(pattern,str(raw.get('operating_system','')),re.I)), 'Linux') retain=policy['retain'] hostname=raw.get('hostname') if raw.get('hostname') in policy['approved_hostnames'] else (hostname_label(raw.get('hostname')) if retain['hostname'] else None) or 'Host-0001' server={'hostname':hostname,'platform':platform, 'operating_system':'Windows Server '+year if year else 'Windows' if platform=='windows' else linux_name, 'os_version':year or numeric_version(raw.get('os_version','')), 'os_build':numeric_version(raw.get('os_build','')), 'last_boot':safe_time(raw.get('last_boot')) or 'Not collected', 'domain':'Domain identity omitted', 'ip_addresses':[ip for value in raw.get('ip_addresses',[]) if (ip:=safe_ip(value))], 'network_adapters':[f'Adapter-{i+1:04d}' for i,_ in enumerate(raw.get('network_adapters',[]))]} if isinstance(raw.get('os_build'),str) and re.fullmatch(r'[0-9]{1,6}',raw['os_build']): server['os_build']=raw['os_build'] if raw.get('product_type') in (1,2,3) and type(raw['product_type']) is int: server['product_type']=raw['product_type'] if type(raw.get('domain_joined')) is bool: server['domain_joined']=raw['domain_joined'] source=trusted_source if trusted_source in SOURCES else None if source: server['collection_source']=source result={'schema_version':'1.0','synthetic':data.get('synthetic') is True,'collected_at':stamp, 'server':server,'checks':{},'inventory':{key:[] for key in ('roles','services','network','tasks','software','storage','certificates')}, 'privacy':{'profile':'minimized-metadata-v1','identities_omitted':True,'paths':'share-and-mount-roots-only', 'retained':sorted(key for key,on in retain.items() if on), 'scope':'Disclosure reduction only; not a PHI-free or compliance attestation.'}} own={ipaddress.ip_address(ip) for ip in server['ip_addresses']} for section in result['inventory']: rows=data['inventory'].get(section,[]) if not isinstance(rows,list) or len(rows)>20000: raise ValueError('Inventory section exceeds its permitted structure or row limit.') for i,row in enumerate(rows): if not isinstance(row,dict): raise ValueError('Inventory row must be an object.') name=row.get('name','') if section=='roles': normalized=str(name).lower() kept=normalized if normalized in KNOWN_ROLES else (safe_label(name) or '').lower() if retain['service_names'] else '' result['inventory'][section].append({'name':kept or f'Component-{i+1:04d}','state':'Installed' if row.get('state')=='Installed' else 'Unknown'}) elif section=='services': result['inventory'][section].append({'name':service_label(name,i,policy),'state':enum_value(row.get('state'), SERVICE_STATES), 'startup':enum_value(row.get('startup'), STARTUP),'pid':integer(row.get('pid')),'account':OMITTED}) elif section=='network': process=normal_name(row.get('process')) scope,peer=remote_scope(row.get('remote_address'),own,row.get('remote_scope'),retain['private_peer_addresses']) result['inventory'][section].append({'state':enum_value(row.get('state'), NETWORK_STATES), 'local_address':safe_ip(row.get('local_address')),'remote_address':peer,'remote_scope':scope, 'local_port':integer(row.get('local_port'),65535),'remote_port':integer(row.get('remote_port'),65535), 'pid':integer(row.get('pid')),'process':process if process in KNOWN_PROCESSES or (retain['process_names'] and safe_label(process)) else f'Process-{i+1:04d}', 'direction':row.get('direction') if row.get('direction') in ('Inbound','Outbound','Unknown') else 'Unknown'}) elif section=='tasks': routine=task_is_maintenance(row) kept_name=safe_label(name) if retain['task_names'] else None kept_path=safe_path(row.get('path')) if retain['task_names'] else None if kept_path is None and retain['task_names'] and row.get('path')=='systemd system timer': kept_path='systemd system timer' result['inventory'][section].append({'name':kept_name or (f'Maintenance-{i+1:04d}' if routine else f'Task-{i+1:04d}'), 'path':kept_path or ('\\Microsoft\\Windows\\Maintenance\\' if routine else 'Task folder omitted'), 'enabled':row.get('enabled') is True,'state':enum_value(row.get('state'), TASK_STATES), 'account':OMITTED,'triggers':'Schedule details omitted; verify through an authorized scheduler review.'}) elif section=='software': label=name if isinstance(name,str) and name in policy['approved_software_names'] else None if label is None and retain['software_names']: label=safe_label(name) if label is None: label=next((product for product in PRODUCTS if isinstance(name,str) and re.match(re.escape(product)+r'(?![A-Za-z0-9])',name,re.I)),f'Software-{i+1:04d}') result['inventory'][section].append({'name':label,'version':numeric_version(row.get('version'))}) elif section=='storage': kind=row.get('type') if row.get('type') in ('Volume','SMB Share','NFS Export','Samba Share','Remote Mount','Mounted Filesystem') else 'Volume' admin=isinstance(name,str) and re.fullmatch(r'(?:ADMIN|IPC|[A-Z])\$',name,re.I) path=OMITTED try: canonical=canonical_share_root(row.get('path')) approved={canonical_share_root(root) for root in policy['approved_share_roots']} if canonical in approved: path=row['path'] except ValueError: pass keep=retain['storage_names_and_roots'] if path==OMITTED and keep and safe_path(row.get('path')): path=safe_path(row.get('path')) shown=(safe_label(name) or safe_path(name)) if keep else None entry={'type':kind,'name':name.upper() if admin else shown if shown else f'Share-{i+1:04d}' if kind in ('SMB Share','NFS Export','Samba Share') else f'Volume-{i+1:04d}', 'path':path,'capacity_gib':None,'free_gib':None} for key in ('capacity_gib','free_gib'): number=row.get(key) if type(number) in (int,float) and 0<=number<=10**12: entry[key]=number if row.get('filesystem') in ('NTFS','ReFS','FAT32','exFAT','ext4','xfs','btrfs','tmpfs','overlay'): entry['filesystem']=row['filesystem'] if type(row.get('special')) is bool: entry['special']=row['special'] # Historical aggregate observations can be retained without paths/identities. if row.get('activity_source') in ('Get-SmbOpenFile (share-root-correlated)','Get-SmbOpenFile (administrative alias ambiguous)'): entry['activity_source']=row['activity_source'] for key in ('observed_open_files','observed_client_count'): entry[key]=integer(row.get(key)) observed=safe_time(row.get('activity_observed_at')) if observed: entry['activity_observed_at']=observed result['inventory'][section].append(entry) elif section=='certificates': entry={'subject':f'Certificate-{i+1:04d}','issuer':OMITTED} if safe_time(row.get('expires')): entry['expires']=safe_time(row['expires']) if row.get('store') in ('My','Root','CA','TrustedPeople','TrustedPublisher','AuthRoot','Disallowed'): entry['store']=row['store'] result['inventory'][section].append(entry) for name,check in data['checks'].items(): if name not in SECTIONS: continue if not isinstance(check,dict): raise ValueError('Invalid collection check.') status=enum_value(check.get('status'), SCORES, 'failed') clean={'status':status,'message':{'complete':'Documented query scope completed.','partial':'Scope incomplete; manual review required.','failed':'Collection check failed; raw diagnostics omitted.','not_collected':'Scope not collected; review required.'}[status], 'errors':[],'permission_limitations':[],'limitations':['Identifying text is minimized; metadata may have entered source-process memory before projection.']} for field in ('started_at','completed_at'): if safe_time(check.get(field)): clean[field]=safe_time(check[field]) for error in (check.get('errors',[]) if isinstance(check.get('errors',[]),list) else [])[:30]: code=error.get('code') if isinstance(error,dict) else 'collection_failed' if not isinstance(code,str) or code not in ERROR_CODES: code='collection_failed' clean['errors'].append({'code':code,'scope':name,'message':'Collection limitation; raw diagnostics omitted.'}) if check.get('permission_limitations'): clean['permission_limitations']=['Current token may have incomplete read access.'] result['checks'][name]=clean result['checks']['privacy']={'status':'partial','message':'Identifying fields omitted. Source metadata may enter memory; this mode is not a PHI-free attestation.', 'errors':[],'permission_limitations':[],'limitations':['Names/accounts/path detail are reduced. Numeric IP addresses remain infrastructure-confidential data. Legacy imports are processed before projection.']} return result