76 lines
2.8 KiB
Python
Executable File
76 lines
2.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import json
|
|
import pathlib
|
|
import socket
|
|
import time
|
|
|
|
SOCKET = '/run/aios-governance-executor/executor.sock'
|
|
STATE = pathlib.Path('/var/lib/aios-gitops/governance-executor/drift-monitor.json')
|
|
APPROVED = pathlib.Path('/var/lib/aios-gitops/runtime/tasks/AIOS-PROTECTION-SNAPSHOT-DRIFT-FORENSICS-V1/protection-drift/semantic-analysis.json')
|
|
|
|
|
|
def call():
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.settimeout(20)
|
|
sock.connect(SOCKET)
|
|
request = {'operation': 'GET_BRANCH_PROTECTION', 'payload': {'repository': 'aios/aios', 'branch': 'main'}}
|
|
sock.sendall((json.dumps(request, separators=(',', ':')) + '\n').encode())
|
|
chunks = []
|
|
while True:
|
|
part = sock.recv(1024 * 1024)
|
|
if not part:
|
|
break
|
|
chunks.append(part)
|
|
if b'\n' in part:
|
|
break
|
|
sock.close()
|
|
return json.loads(b''.join(chunks).splitlines()[0])
|
|
|
|
|
|
def main():
|
|
try:
|
|
result = call()
|
|
if result.get('status') != 'PASS':
|
|
raise RuntimeError(str(result.get('code') or 'EXECUTOR_GET_FAILED'))
|
|
approved = json.loads(APPROVED.read_text(encoding='utf-8'))
|
|
approved_transport = approved['approved_transport_digest_from_evidence']
|
|
approved_semantic = approved['approved_semantic_digest']
|
|
transport = result['transport_digest']
|
|
semantic = result['semantic_digest']
|
|
if semantic != approved_semantic:
|
|
state = 'GOVERNANCE_POLICY_DRIFT'
|
|
elif transport != approved_transport:
|
|
state = 'GOVERNANCE_TRANSPORT_METADATA_CHANGED'
|
|
else:
|
|
state = 'NO_DRIFT'
|
|
payload = {
|
|
'schema': 'aios.governance-drift-monitor.v1',
|
|
'observed_at_unix': int(time.time()),
|
|
'status': state,
|
|
'release_blocking': False,
|
|
'approved_transport_digest': approved_transport,
|
|
'observed_transport_digest': transport,
|
|
'approved_semantic_digest': approved_semantic,
|
|
'observed_semantic_digest': semantic,
|
|
'semantic_schema': result.get('semantic_schema'),
|
|
'source': 'Dedicated Governance Executor GET_BRANCH_PROTECTION',
|
|
}
|
|
except Exception as exc:
|
|
payload = {
|
|
'schema': 'aios.governance-drift-monitor.v1',
|
|
'observed_at_unix': int(time.time()),
|
|
'status': 'MONITOR_UNAVAILABLE',
|
|
'release_blocking': False,
|
|
'error_code': str(exc),
|
|
}
|
|
STATE.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = STATE.with_suffix('.tmp')
|
|
temporary.write_text(json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=False) + '\n', encoding='utf-8')
|
|
temporary.replace(STATE)
|
|
STATE.chmod(0o640)
|
|
print(json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=False))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|