NixDaemon

NixOS pentest workstation as one flake — IceBreaker's successor
git clone https://git.daemon-sec.xyz/NixDaemon.git
Log | Files | Refs | README

stability_guard.py (4484B)


      1 #!/usr/bin/python3
      2 """Use ordinary Linux CPU ceilings to reduce power/temperature clamp triggers."""
      3 from datetime import datetime, timezone
      4 import json
      5 import math
      6 import os
      7 from pathlib import Path
      8 import signal
      9 import time
     10 
     11 POLICIES = Path('/sys/devices/system/cpu/cpufreq')
     12 STATUS = Path('/run/motherboard-stability/status.json')
     13 
     14 
     15 def read_number(path, divisor=1):
     16     try:
     17         value = float(Path(path).read_text().strip()) / divisor
     18         return value if math.isfinite(value) else None
     19     except (OSError, ValueError):
     20         return None
     21 
     22 
     23 def choose_limit(ac_online, cpu_c, gpu_c, main_rpm, previous_stage=0):
     24     if (cpu_c is None or gpu_c is None or main_rpm is None or main_rpm <= 0
     25             or not all(math.isfinite(t) and 0 <= t <= 125 for t in (cpu_c, gpu_c))):
     26         return 1800000, 2, 'essential sensor or main fan unavailable'
     27     hottest = max(cpu_c, gpu_c)
     28     if hottest >= 73 or (previous_stage == 2 and hottest > 66):
     29         return 1800000, 2, 'hot: CPU or board GPU temperature at/above 73C, or cooling down'
     30     if hottest >= 65 or (previous_stage >= 1 and hottest > 58):
     31         return 2400000, 1, 'warm: CPU or board GPU temperature at/above 65C, or cooling down'
     32     if ac_online is not True:
     33         return 2400000, 0, 'battery or mains status unavailable'
     34     return 3200000, 0, 'mains power; normal temperatures'
     35 
     36 
     37 def find_hwmon(name):
     38     for path in Path('/sys/class/hwmon').glob('hwmon*'):
     39         try:
     40             if (path / 'name').read_text().strip() == name:
     41                 return path
     42         except OSError:
     43             continue
     44     return None
     45 
     46 
     47 def set_limits(target):
     48     policies = sorted(POLICIES.glob('policy*'))
     49     if len(policies) != 16:
     50         raise RuntimeError('Unexpected CPU policy count')
     51     for p in policies:
     52         low = read_number(p / 'scaling_min_freq')
     53         high = read_number(p / 'cpuinfo_max_freq')
     54         if low is None or high is None or not low <= target <= high:
     55             raise RuntimeError('Requested ceiling conflicts with CPU bounds: ' + p.name)
     56     for p in policies:
     57         if read_number(p / 'scaling_max_freq') != target:
     58             (p / 'scaling_max_freq').write_text(str(target))
     59     deadline = time.monotonic() + 5
     60     while True:
     61         values = [read_number(p / 'scaling_max_freq') for p in policies]
     62         if all(value == target for value in values):
     63             return
     64         if time.monotonic() >= deadline:
     65             raise RuntimeError('CPU ceiling readback did not settle: ' + repr(values))
     66         time.sleep(0.05)
     67 
     68 
     69 def write_status(data):
     70     temporary = STATUS.with_suffix('.tmp')
     71     temporary.write_text(json.dumps(data, indent=2) + '\n')
     72     os.replace(temporary, STATUS)
     73 
     74 
     75 def main():
     76     if Path('/sys/class/dmi/id/board_name').read_text().strip() != 'GM7RGxM':
     77         raise RuntimeError('Unsupported motherboard')
     78     if 'AMD Ryzen 9 6900HX' not in Path('/proc/cpuinfo').read_text():
     79         raise RuntimeError('Unsupported CPU')
     80     stage = 0
     81     running = True
     82     last = None
     83     data = {}
     84     def stop(signum, frame):
     85         nonlocal running
     86         running = False
     87     signal.signal(signal.SIGTERM, stop)
     88     signal.signal(signal.SIGINT, stop)
     89     try:
     90         while running:
     91             cpu, board = find_hwmon('k10temp'), find_hwmon('uniwill')
     92             cpu_c = read_number(cpu / 'temp1_input', 1000) if cpu else None
     93             gpu_c = read_number(board / 'temp2_input', 1000) if board else None
     94             rpm = read_number(board / 'fan1_input') if board else None
     95             ac = read_number('/sys/class/power_supply/AC0/online') == 1
     96             target, stage, reason = choose_limit(ac, cpu_c, gpu_c, rpm, stage)
     97             set_limits(target)
     98             data = {'running': True, 'timestamp': datetime.now(timezone.utc).isoformat(),
     99                     'limit_mhz': target // 1000, 'thermal_stage': stage, 'reason': reason,
    100                     'mains_online': ac, 'cpu_c': cpu_c, 'board_gpu_c': gpu_c, 'main_fan_rpm': rpm,
    101                     'note': 'Avoidance workaround; firmware thermal/power protection remains active.'}
    102             write_status(data)
    103             if last != (target, stage, reason):
    104                 print(json.dumps(data), flush=True)
    105                 last = (target, stage, reason)
    106             time.sleep(1)
    107     finally:
    108         if data:
    109             data['running'] = False
    110             data['timestamp'] = datetime.now(timezone.utc).isoformat()
    111             write_status(data)
    112 
    113 
    114 if __name__ == '__main__':
    115     main()