fix(monitor): keep Docker web links and schemes accurate

- Reuse the Find snapshot for web links only after a complete search
- Skip unassigned or ranged host ports instead of dropping the container
- Prefill the https scheme of published HTTPS ports in the editor
- Run the Docker discovery tests when they change
- Translate the Docker search texts
This commit is contained in:
MacRimi
2026-10-19 17:12:28 +02:00
parent e44db4897b
commit 7bf1e1e34d
9 changed files with 95 additions and 14 deletions
+15 -9
View File
@@ -116,18 +116,19 @@ def public_ports(record):
if not isinstance(value, dict):
raise ValueError('invalid binding')
address = value.get('HostIp') or '0.0.0.0'
raw = value.get('HostPort')
# An unassigned ("") or ranged host port is not a single address.
if not isinstance(raw, str) or not raw.isdecimal():
continue
try:
ip = ipaddress.ip_address(address)
if ip.is_loopback or getattr(ip, 'ipv4_mapped', None) and ip.ipv4_mapped.is_loopback:
continue
raw = value.get('HostPort')
if not isinstance(raw, str) or not raw.isdecimal():
raise ValueError('invalid port')
outer = int(raw)
except (ValueError, TypeError):
raise ValueError('invalid binding') from None
continue
if ip.is_loopback or getattr(ip, 'ipv4_mapped', None) and ip.ipv4_mapped.is_loopback:
continue
outer = int(raw)
if not 1 <= outer <= 65535:
raise ValueError('invalid port')
continue
result[outer] = {'host_port': outer, 'container_port': inner,
'scheme': 'https' if inner in {443, 4443, 8443, 9443} or outer in {443, 4443, 8443, 9443} else 'http'}
return sorted(result.values(), key=lambda item: item['host_port'])
@@ -167,6 +168,9 @@ def _candidate(vmid, record):
'container_name': record['name'], 'name': name,
'installed_version': record['version'], 'evidence': evidence,
'default_ports': [port['host_port'] for port in record['endpoints'] if port['container_port'] in WEB_PORTS],
'port_details': [{'port': port['host_port'], 'scheme': port['scheme'], 'path': None,
'description': '', 'logo_url': ''}
for port in record['endpoints'] if port['container_port'] in WEB_PORTS] or None,
'tracking_suggestion': None}
@@ -272,7 +276,9 @@ def cached_web_links(vmid, lookup):
"""Reuse a Find snapshot in existing web-link suggestions without another scan."""
with _mutex:
value = copy.deepcopy(_cache.get(int(vmid)))
if not value or value.get('_retired') or time.time() - value['_attempt_at'] >= SNAPSHOT_SECONDS:
# Only a complete search lists every container; otherwise probe as before.
if (not value or value.get('_retired') or value.get('status') != 'complete'
or time.time() - value['_attempt_at'] >= SNAPSHOT_SECONDS):
return None
result = []
for record in value.get('_containers', []):
@@ -100,6 +100,21 @@ class Scan(unittest.TestCase):
self.assertEqual(item['default_ports'],[8080])
self.assertIn({'kind':'tcp_listener','value':'21115'},item['evidence'])
def test_unassigned_or_ranged_host_ports_are_skipped_not_invalid(self):
app = record(running=False, bindings={'8080/tcp': [{'HostIp': '', 'HostPort': ''}],
'9000/tcp': [{'HostIp': '', 'HostPort': '9000-9010'}],
'443/tcp': [{'HostIp': '', 'HostPort': '8444'}]})
result = self.scan(records=[app])
self.assertEqual(result['status'], 'complete')
endpoints = result['_containers'][0]['endpoints']
self.assertEqual([e['host_port'] for e in endpoints], [8444])
def test_https_container_port_prefills_its_scheme(self):
app = record(bindings={'443/tcp': [{'HostIp': '0.0.0.0', 'HostPort': '8444'}]})
item, = self.scan(records=[app])['candidates']
self.assertEqual(item['default_ports'], [8444])
self.assertEqual([(d['port'], d['scheme']) for d in item['port_details']], [(8444, 'https')])
def test_empty_engine_is_complete_and_missing_docker_never_calls_docker(self):
self.assertEqual(self.scan(ids=[])['status'], 'complete')
self.calls.clear()
@@ -202,6 +217,20 @@ class Cache(unittest.TestCase):
self.assertEqual(lxc_apps._probe_docker_web_links(101)[0]['host_port'], 8080)
execute.assert_not_called()
def test_incomplete_search_never_replaces_the_web_link_probe(self):
lookup = Mock(return_value={'slug': None, 'name': 'web', 'logo_url': None})
partial = {**copy.deepcopy(self.value), 'status': 'partial', 'issues': ['inspect_failed']}
with patch.object(docker, 'scan', return_value=partial), patch.object(docker.time, 'time', return_value=1000):
docker.discover(101)
with patch.object(docker.time, 'time', return_value=1001):
self.assertIsNone(docker.cached_web_links(101, lookup))
docker._cache.clear()
with patch.object(docker, 'scan', side_effect=native.ProbeError('timeout')), \
patch.object(docker.time, 'time', return_value=1000):
docker.discover(101)
with patch.object(docker.time, 'time', return_value=1001):
self.assertIsNone(docker.cached_web_links(101, lookup))
def test_shared_links_keep_existing_name_then_port_order(self):
value = copy.deepcopy(self.value)
value['_containers'][0]['name'] = 'z-last'