fix: recognize Hestia panel and filter watch helpers

This commit is contained in:
VAIO73
2026-10-17 11:27:16 +00:00
parent e978bac91a
commit 1f75147d51
2 changed files with 43 additions and 3 deletions
+25 -3
View File
@@ -74,8 +74,11 @@ _PACKAGE_ROOT_RUNTIMES = {'node', 'nodejs', 'npm', 'pnpm', 'yarn', 'bun'}
_HELPER_ROLES = {
'backup', 'bootstrap', 'bridge', 'cleanup', 'housekeeping', 'maintenance',
'migrate', 'migration', 'restore', 'rotate', 'update', 'updater', 'upgrade',
'watchdog', 'worker',
'watch', 'watchdog', 'worker',
}
_HESTIA_NGINX = '/usr/local/hestia/nginx/sbin/hestia-nginx'
_HESTIA_ROOT = '/usr/local/hestia'
_HESTIA_PANEL_PORT = 8083
class ProbeError(Exception):
@@ -253,6 +256,17 @@ def _helper(name):
return any(part in _HELPER_ROLES for part in re.split(r'[._-]+', raw) if part)
def _hestia_panel_process(executable, ports):
"""Recognize only Hestia's own bound control-panel process.
Hestia's generated init unit does not expose an ExecStart that the generic
service parser may treat as an application. Its panel process and port are
stable, local runtime metadata, so this stays narrower than accepting
arbitrary vendor services or a generic web server.
"""
return executable == _HESTIA_NGINX and _HESTIA_PANEL_PORT in ports
def _custom(path):
if not _path(path):
return False
@@ -367,7 +381,10 @@ def parse_records(text, kind):
records.append(record)
elif kind == 'P' and len(parts) == 4 and parts[0] == 'P':
_, pid, exe, directory = parts
if not pid.isdigit() or not _path(exe) or not _path(directory):
# A process rooted at / is normally too ambiguous for generic
# discovery, but it is valid process metadata for the exact Hestia
# panel signature below. It never becomes a generic candidate.
if not pid.isdigit() or not _path(exe) or (directory != '/' and not _path(directory)):
continue
records.append({'pid': pid, 'executable': exe, 'root': directory})
elif kind == 'L' and len(parts) == 3 and parts[0] == 'L':
@@ -407,7 +424,8 @@ def candidate(vmid, record, footprint):
evidence.append({'kind': record['manager'], 'value': record['service']})
else:
evidence.extend([{'kind': 'process_executable', 'value': record['process']},
{'kind': 'working_directory', 'value': record['root']}])
{'kind': 'working_directory',
'value': record.get('process_root', record['root'])}])
ports = sorted(set(record.get('listening_ports') or []))
evidence.extend({'kind': 'tcp_listener', 'value': str(port)} for port in ports)
return {'slug': identity, 'discovery_id': identity, 'source': 'native_metadata',
@@ -501,6 +519,10 @@ def scan(vmid, run=bounded_run):
ports = [entry['port'] for entry in listeners if entry['pid'] == process['pid']]
if len(related) == 1:
related[0].setdefault('listening_ports', []).extend(ports)
elif not related and _hestia_panel_process(exe, ports):
records.append({'name': 'Hestia', 'path': exe, 'root': _HESTIA_ROOT,
'process': exe, 'process_root': directory, 'executable': exe,
'listening_ports': ports})
elif not related and (_custom(exe) and _custom(directory)
and not _system(Path(exe).name)
and not _helper(Path(exe).name)
@@ -132,6 +132,7 @@ class DiscoveryFixtures(unittest.TestCase):
('cortendesk-bootstrap-worker', '/bin/bash', '/opt/cortendesk', '/opt/cortendesk/bootstrap-worker.sh'),
('cortendesk-source-worker', '/bin/bash', '/opt/cortendesk', '/opt/cortendesk/source-worker.sh'),
('cortendesk-managed-password@', '/bin/bash', '/opt/cortendesk', '/opt/cortendesk/password-bridge.sh'),
('miminails-watch', '/usr/local/bin/miminails-watch', '/usr/local/bin', ''),
]
for name, exe, directory, script in cases:
with self.subTest(name=name):
@@ -141,6 +142,23 @@ class DiscoveryFixtures(unittest.TestCase):
self.assertEqual(result['candidates'], [])
self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
def test_hestia_panel_is_recognized_only_from_its_exact_runtime_and_port(self):
hestia = '/usr/local/hestia/nginx/sbin/hestia-nginx'
process = f'P\t391\t{hestia}\t/\nEND\n'
result = native.scan(101, self.runner(
services='END\n', processes=process, ports='L\t391\t8083\nEND\n'))
item, = result['candidates']
self.assertEqual(item['name'], 'Hestia')
self.assertEqual(item['installation_path'], hestia)
self.assertEqual(item['listening_ports'], [8083])
self.assertIn({'kind': 'working_directory', 'value': '/'}, item['evidence'])
self.calls.clear()
result = native.scan(101, self.runner(
services='END\n', processes=process, ports='L\t391\t8443\nEND\n'))
self.assertEqual(result['candidates'], [])
self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
def test_node_and_npm_wrappers_use_only_bounded_manifest_roots(self):
cases = [
('keynest', '/usr/bin/node', '/opt/keynest'),