159 lines
7.3 KiB
Python
159 lines
7.3 KiB
Python
"""Metadata-only file checks and local Ubuntu update evidence.
|
|
|
|
No privilege escalation, package refresh, network calls, private-file reads or
|
|
automatic repairs. Findings are review prompts, never exploit confirmations.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
import os
|
|
from pathlib import Path
|
|
import platform
|
|
import re
|
|
import stat
|
|
import time
|
|
|
|
from .reports import utc_now
|
|
|
|
|
|
POLICIES = (
|
|
('/etc/passwd',False), ('/etc/group',False),
|
|
('/etc/shadow',True), ('/etc/gshadow',True),
|
|
('/etc/sudoers',False), ('/etc/ssh/sshd_config',False),
|
|
('/etc',False), ('/etc/ssh',False), ('/etc/sudoers.d',False),
|
|
)
|
|
|
|
|
|
def inspect_path(path, sensitive=False, *, lstat=os.lstat):
|
|
row = {'source':path,'method':'lstat_metadata_only','status':'unknown',
|
|
'observed_at':utc_now(),'reasons':[]}
|
|
try:
|
|
info = lstat(path)
|
|
except FileNotFoundError:
|
|
row.update(status='skipped',error_code='PATH_MISSING')
|
|
return row
|
|
except OSError:
|
|
row['error_code'] = 'METADATA_UNAVAILABLE'
|
|
return row
|
|
mode = stat.S_IMODE(info.st_mode)
|
|
row.update(mode=oct(mode),uid=info.st_uid,gid=info.st_gid)
|
|
if stat.S_ISLNK(info.st_mode):
|
|
row['error_code'] = 'SYMLINK_REQUIRES_REVIEW'
|
|
return row
|
|
if not (stat.S_ISREG(info.st_mode) or stat.S_ISDIR(info.st_mode)):
|
|
row['error_code'] = 'UNEXPECTED_FILE_TYPE'
|
|
return row
|
|
if mode & stat.S_IWOTH:
|
|
row['reasons'].append('WORLD_WRITABLE')
|
|
if mode & stat.S_IWGRP:
|
|
row['reasons'].append('GROUP_WRITABLE_REVIEW')
|
|
if info.st_uid != 0:
|
|
row['reasons'].append('OWNER_NOT_ROOT_REVIEW')
|
|
if sensitive and mode & stat.S_IROTH:
|
|
row['reasons'].append('SENSITIVE_WORLD_READABLE')
|
|
row['status'] = 'review' if row['reasons'] else 'observed'
|
|
row['limitations'] = ['POSIX mode/owner only; ACLs, capabilities, mount options and runtime use not evaluated.',
|
|
'Group access or non-root ownership can be intentional; review in deployment context.']
|
|
return row
|
|
|
|
|
|
def read_small(path):
|
|
try:
|
|
with Path(path).open(encoding='utf-8') as stream:
|
|
value = stream.read(8193)
|
|
return value if len(value) <= 8192 else None
|
|
except (OSError,UnicodeError):
|
|
return None
|
|
|
|
|
|
def update_evidence(*, lstat=os.lstat, read=read_small, now=time.time):
|
|
"""A timestamp is freshness evidence, never a vulnerability inventory."""
|
|
row = {'method':'local_update_cache_only','observed_at':utc_now(),'status':'unknown',
|
|
'pending_updates':None,'security_updates':None,'sources':[],
|
|
'limitations':['No repository refresh or package solver was run.',
|
|
'Missing or stale cache cannot establish whether the system is fully patched.']}
|
|
for path in ('/var/lib/apt/periodic/update-success-stamp',
|
|
'/var/lib/update-notifier/updates-available'):
|
|
evidence = {'source':path,'status':'unknown'}
|
|
try:
|
|
info = lstat(path)
|
|
if not stat.S_ISREG(info.st_mode):
|
|
evidence['error_code'] = 'NOT_REGULAR_FILE'
|
|
else:
|
|
timestamp = info.st_mtime
|
|
age = now()-timestamp
|
|
if age < 0:
|
|
evidence['error_code'] = 'FUTURE_TIMESTAMP'
|
|
else:
|
|
evidence.update(status='observed',age_hours=round(age/3600,2),
|
|
modified_at=datetime.fromtimestamp(timestamp,timezone.utc).isoformat())
|
|
if path.endswith('updates-available'):
|
|
# This notifier summary is optional and may be localized.
|
|
# Never retain its complete text in logs or reports.
|
|
text = read(path)
|
|
match = re.search(r'^\s*(\d+) updates? can be applied immediately\.',text or '',re.M)
|
|
if match:
|
|
row['pending_updates'] = int(match[1])
|
|
row['status'] = 'cached_observation'
|
|
match = re.search(r'^(\d+) of these updates? (?:is|are) (?:a )?(?:standard )?security updates?\.',text or '',re.M)
|
|
if match:
|
|
row['security_updates'] = int(match[1])
|
|
except FileNotFoundError:
|
|
evidence['error_code'] = 'CACHE_MISSING'
|
|
except (OSError,ValueError,OverflowError):
|
|
evidence['error_code'] = 'CACHE_UNAVAILABLE'
|
|
row['sources'].append(evidence)
|
|
return row
|
|
|
|
|
|
def collect(report=None, on_result=None, *, system=None, lstat=os.lstat, read=read_small, now=time.time):
|
|
report = {} if report is None else report
|
|
report.update(schema_version=1,observed_at=utc_now(),status='unsupported',files=[],updates=None,
|
|
limitations=['Read-only hints, not a complete security audit.',
|
|
'Cloud security groups and external reachability cannot be confirmed locally.'])
|
|
if (system or platform.system()) != 'Linux':
|
|
return report
|
|
for path,sensitive in POLICIES:
|
|
report['files'].append(inspect_path(path,sensitive,lstat=lstat))
|
|
if on_result is not None:
|
|
on_result()
|
|
report['updates'] = update_evidence(lstat=lstat,read=read,now=now)
|
|
report['status'] = 'partial'
|
|
report['review_count'] = sum(item['status']=='review' for item in report['files'])
|
|
if on_result is not None:
|
|
on_result()
|
|
return report
|
|
|
|
|
|
def render(report, out):
|
|
tr = out.text
|
|
out.row(tr('权限检查范围','Permission scope'),tr('只读权限元数据','Read-only metadata'),'unknown')
|
|
reasons = {
|
|
'WORLD_WRITABLE':('所有用户可写','Writable by everyone'),
|
|
'GROUP_WRITABLE_REVIEW':('同组可写，需审阅','Group-writable; review'),
|
|
'OWNER_NOT_ROOT_REVIEW':('所有者非 root，需审阅','Non-root owner; review'),
|
|
'SENSITIVE_WORLD_READABLE':('敏感文件所有用户可读','Sensitive file readable by everyone'),
|
|
}
|
|
for row in report['files']:
|
|
if row['status'] == 'review':
|
|
value = '; '.join(tr(*reasons[reason]) for reason in row['reasons'])
|
|
elif row['status'] == 'observed':
|
|
value = tr('模式位未见上述宽松项','No listed mode-bit finding')
|
|
else:
|
|
value = tr('未知/跳过','Unknown/skipped')+' ('+row.get('error_code','UNKNOWN')+')'
|
|
out.row(row['source'],row.get('mode','---')+' · '+value,
|
|
'bad' if row['status']=='review' else 'ok' if row['status']=='observed' else 'unknown')
|
|
update = report.get('updates')
|
|
if update:
|
|
pending = update['pending_updates']
|
|
out.row(tr('缓存待更新数','Cached pending updates'),tr('未知','Unknown') if pending is None else str(pending),'unknown')
|
|
for source in update['sources']:
|
|
if 'age_hours' in source:
|
|
out.row(tr('更新缓存年龄','Update cache age'),f"{source['age_hours']:g} h · "+source['source'],'unknown')
|
|
out.row(tr('安全结论','Security conclusion'),tr('本机只读观测；补丁状态未确认','Local read-only observations; patch status unconfirmed'),'unknown')
|
|
|
|
|
|
def run_console(lang='zh', color='auto', report=None, on_result=None):
|
|
from .console import Console
|
|
report = collect(report,on_result)
|
|
render(report,Console(lang,color))
|
|
return 1 if report.get('review_count') or report['status'] == 'unsupported' else 0
|