oceans2alaska
610a3bb447
- Add privacy projection (core/privacy.py), private report storage, bounded
process capture; omit identities, free text and unapproved paths
- Mask remote peer IPs: only loopback/host/wildcard stay numeric, others are
reduced to remote_scope private/public (schema + scoring updated)
- Map ss TCP states to canonical names; ASCII-only numeric fields; fix
"Microsoft Visual C++" product matching (Python and PowerShell)
- Merge 7fed3b9: -Diagnostic switch, serialization exit code 3, PS7 module
path, UI error detail; keep hardened task/certificate collectors
- Add audit docs, privacy docs and regression tests
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
57 lines
2.8 KiB
Python
57 lines
2.8 KiB
Python
"""Data-only scoring policy. No executable expressions or arbitrary regex rules."""
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
|
||
DEFAULT_POLICY = Path(__file__).resolve().parents[1] / 'assets' / 'scoring_policy.json'
|
||
RULES = {'installed_workload_role', 'running_application_service', 'attributed_application_listener',
|
||
'observed_application_clients', 'configured_business_share', 'observed_smb_transport',
|
||
'observed_business_share_clients', 'configured_custom_schedule', 'verified_domain_controller'}
|
||
|
||
|
||
def load_policy(source=None):
|
||
if source is not None and not isinstance(source,(dict,str,Path)):
|
||
raise ValueError('A scoring policy must be a JSON object or file path.')
|
||
if isinstance(source, dict):
|
||
policy = copy.deepcopy(source)
|
||
else:
|
||
path = Path(source) if source is not None else DEFAULT_POLICY
|
||
try:
|
||
if path.stat().st_size > 65536:
|
||
raise ValueError('Scoring policy exceeds 64 KiB.')
|
||
policy = json.loads(path.read_text(encoding='utf-8'))
|
||
except (OSError, UnicodeError, json.JSONDecodeError, RecursionError):
|
||
raise ValueError('Unable to read a valid UTF-8 scoring policy JSON file.') from None
|
||
try:
|
||
limits, weights = policy['thresholds'], policy['coverage_weights']
|
||
valid = (
|
||
set(policy) == {'model_version','name','scores','thresholds','coverage_weights','partial_check_credit'}
|
||
and policy['model_version'] == '2.0' and isinstance(policy['name'],str) and 0<len(policy['name'])<=120
|
||
and set(policy['scores']) == RULES
|
||
and all(type(value) is int and 0<=value<=100 for value in policy['scores'].values())
|
||
and set(limits) == {'low','medium','high','critical'}
|
||
and all(type(value) is int for value in limits.values())
|
||
and 1 == limits['low'] < limits['medium'] < limits['high'] < limits['critical'] <= 100
|
||
and set(weights) == {'system','roles','services','network','tasks','software','storage','shares','certificates'}
|
||
and all(type(weight) is int and 1<=weight<=10 for weight in weights.values())
|
||
and type(policy['partial_check_credit']) in (int,float) and 0<=policy['partial_check_credit']<1
|
||
)
|
||
except (KeyError,TypeError,AttributeError):
|
||
valid = False
|
||
if not valid:
|
||
raise ValueError('Invalid scoring policy. Use the bundled v2 policy structure, integer scores 0–100, ordered thresholds and positive coverage weights.')
|
||
policy['name']='Deployment scoring policy'
|
||
return policy
|
||
|
||
|
||
def policy_id(policy):
|
||
return hashlib.sha256(json.dumps(policy, sort_keys=True).encode('utf-8')).hexdigest()
|
||
|
||
|
||
def risk_level(score, policy):
|
||
for name in ('critical','high','medium','low'):
|
||
if score >= policy['thresholds'][name]:
|
||
return name
|
||
return 'informational'
|