tegwick
fb09e138b8
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09cbc-910e-77a0-b336-3130194350b3
85 lines
4.1 KiB
Python
85 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only SSH inventory; reported synchronization is not a UTC accuracy proof."""
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
|
|
REMOTE = r'''
|
|
import hashlib,json,subprocess
|
|
from datetime import datetime,timezone
|
|
from pathlib import Path
|
|
commands = {
|
|
'clock': ['timedatectl','show','--property=Timezone','--property=NTPSynchronized','--property=NTP','--property=LocalRTC'],
|
|
'upstream': ['timedatectl','show-timesync','--all'],
|
|
'timesyncd': ['systemctl','show','systemd-timesyncd','--property=ActiveState','--property=UnitFileState','--property=FragmentPath','--property=DropInPaths','--property=User'],
|
|
'other_daemons': ['systemctl','is-active','chrony','chronyd','ntp'],
|
|
'virtualization': ['systemd-detect-virt'],
|
|
'ntp_listeners': ['ss','-H','-lun','sport','=',':123'],
|
|
}
|
|
rows={}
|
|
for name,argv in commands.items():
|
|
try:
|
|
result=subprocess.run(argv,capture_output=True,text=True,timeout=10)
|
|
rows[name]={'returncode':result.returncode,'stdout':result.stdout[:32768]}
|
|
except (OSError,subprocess.TimeoutExpired):
|
|
rows[name]={'returncode':None,'error':'command_unavailable_or_timeout'}
|
|
files=[Path('/etc/systemd/timesyncd.conf')]
|
|
for directory in ['/etc/systemd/timesyncd.conf.d','/run/systemd/timesyncd.conf.d','/usr/lib/systemd/timesyncd.conf.d']:
|
|
files.extend(sorted(Path(directory).glob('*.conf')))
|
|
hashes={}
|
|
for p in files:
|
|
try:hashes[str(p)]=hashlib.sha256(p.read_bytes()).hexdigest()
|
|
except OSError:hashes[str(p)]=None
|
|
print(json.dumps({'remote_reported_at':datetime.now(timezone.utc).isoformat(),'commands':rows,'config_sha256':hashes}))
|
|
'''
|
|
|
|
|
|
def observe(host, runner=subprocess.run):
|
|
if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.@-]*', host):
|
|
raise ValueError('host must be an SSH alias or hostname, not options or shell text')
|
|
result = runner(['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10',
|
|
host, 'python3 -'], input=REMOTE, text=True,
|
|
capture_output=True, timeout=90)
|
|
if result.returncode != 0:
|
|
raise ValueError('SSH inventory failed; no usable inventory was produced')
|
|
if len(result.stdout.encode()) > 256 * 1024:
|
|
raise ValueError('inventory exceeded size limit')
|
|
data = json.loads(result.stdout)
|
|
if not isinstance(data, dict) or not isinstance(data.get('commands'), dict):
|
|
raise ValueError('invalid inventory shape')
|
|
clock = data['commands'].get('clock', {})
|
|
properties = dict(line.split('=', 1) for line in clock.get('stdout', '').splitlines() if '=' in line)
|
|
return {'schema_version': 1, 'host_alias': host,
|
|
'collector_reported_at': datetime.now(timezone.utc).isoformat(),
|
|
'read_only': True, 'authority_usable': None,
|
|
'synchronized_reported': {'yes': True, 'no': False}.get(properties.get('NTPSynchronized')) if clock.get('returncode') == 0 else None,
|
|
'limitations': ['OS synchronization flag is not an independent UTC error bound',
|
|
'host and collector timestamps are observations, not authenticated time samples',
|
|
'SSH uses existing caller identity/trust; no sudo, installation or time mutation'],
|
|
'collector_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
|
|
'remote_script_sha256': hashlib.sha256(REMOTE.encode()).hexdigest(),
|
|
'inventory': data}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('host')
|
|
parser.add_argument('--output', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
result = observe(args.host)
|
|
# Refuse overwriting historical observations.
|
|
with args.output.open('x') as stream:
|
|
json.dump(result, stream, indent=2)
|
|
stream.write('\n')
|
|
except (OSError, ValueError, subprocess.TimeoutExpired):
|
|
parser.exit(1, 'Inventory failed or output already exists; no success claimed.\n')
|
|
print(f'Inventory saved to {args.output}; UTC accuracy remains unverified.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|