fix(monitor): isolate local icon fetching and image decoding

AppImage/scripts/local_app_icons.py

@@ -1,6 +1,7 @@
 """Explicit LXC icon lookup. No guest execution, background scan or app writes.
 
-Untrusted HTTP and image parsing run in a bounded child. Cache objects are
+Untrusted HTTP and image parsing run in an unprivileged bounded child. Only
+the target CT's pinned network namespace may be used for HTTP. Cache objects are
 immutable: fetching a draft cannot overwrite a registered application's icon.
 """
 from __future__ import annotations
@@ -15,6 +16,7 @@ import json
 import os
 from pathlib import Path
 import re
+import signal
 import socket
 import ssl
 import subprocess
@@ -30,6 +32,7 @@ from html.parser import HTMLParser
 CACHE = Path('/etc/proxmenux/apps/logo-cache')
 APPS = Path('/etc/proxmenux/apps')
 CONFIG = Path('/etc/pve/lxc')
+PROC = Path('/proc')
 MAX_BYTES = 256 * 1024
 MAX_ENTRIES = 32
 MAX_DIMENSION = 2048
@@ -37,10 +40,13 @@ MAX_PIXELS = 1024 * 1024
 WORKER_TIMEOUT = 8
 UNUSED_TTL = 24 * 3600
 SLOTS = threading.BoundedSemaphore(4)
+DECODERS = threading.BoundedSemaphore(4)
 TOKEN = re.compile(r'^[a-f0-9]{32}$')
 URL = re.compile(r'^/api/vms/([1-9][0-9]*)/apps/logo-cache/([a-f0-9]{32})$')
 MIMES = {'image/png', 'image/jpeg', 'image/gif', 'image/webp',
          'image/x-icon', 'image/vnd.microsoft.icon', 'image/svg+xml'}
+RASTER_FORMATS = {'image/png': 'PNG', 'image/jpeg': 'JPEG', 'image/gif': 'GIF',
+                 'image/webp': 'WEBP', 'image/x-icon': 'ICO', 'image/vnd.microsoft.icon': 'ICO'}
 
 
 class IconError(ValueError):
@@ -106,6 +112,62 @@ def _origin(vmid, payload):
     return f"{payload['scheme']}://{'[' + host + ']' if ':' in host else host}:{payload['port']}"
 
 
+def _init_pid(vmid):
+    value = _run(['lxc-info', '-n', str(vmid), '-pH']).strip()
+    if not re.fullmatch(r'[1-9][0-9]{0,9}', value) or int(value) <= 1:
+        raise IconError('container init process unavailable')
+    return int(value)
+
+
+def _process_start(pid):
+    # comm may contain spaces and parentheses; field 22 follows the last ')'.
+    value = (PROC / str(pid) / 'stat').read_text().rsplit(')', 1)[1].split()[19]
+    if not value.isdecimal():
+        raise IconError('container process identity unavailable')
+    return value
+
+
+def _ns_id(info):
+    return info.st_dev, info.st_ino
+
+
+def _network_unchanged(vmid, identity):
+    pid, started, namespace = identity
+    return (_init_pid(vmid) == pid and _process_start(pid) == started
+            and _ns_id((PROC / str(pid) / 'ns/net').stat()) == namespace)
+
+
+@contextlib.contextmanager
+def _network(vmid):
+    """Pin nsfs before launching; revalidate PID/start time to reject PID reuse."""
+    fd = None
+    try:
+        pid = _init_pid(vmid)
+        started = _process_start(pid)
+        fd = os.open(PROC / str(pid) / 'ns/net', os.O_RDONLY | os.O_CLOEXEC)
+        identity = pid, started, _ns_id(os.fstat(fd))
+        if (identity[2] == _ns_id((PROC / 'self/ns/net').stat())
+                or not _network_unchanged(vmid, identity)):
+            raise IconError('container network is not isolated or changed')
+        yield fd, identity
+    except (OSError, IndexError, ValueError) as exc:
+        raise IconError('container network unavailable') from exc
+    finally:
+        if fd is not None:
+            os.close(fd)
+
+
+def _owns_address(value):
+    """Called in the CT network namespace, never trust lxc-info as a target."""
+    try:
+        interfaces = json.loads(_run(['/usr/sbin/ip', '-json', 'address', 'show']))
+        owned = {str(ipaddress.ip_address(a['local'])) for interface in interfaces
+                 for a in interface.get('addr_info', []) if 'local' in a}
+        return str(ipaddress.ip_address(value)) in owned
+    except (ValueError, TypeError, KeyError, AttributeError) as exc:
+        raise IconError('container address ownership unavailable') from exc
+
+
 def _same_origin(left, right):
     try:
         a, b = urllib.parse.urlsplit(left), urllib.parse.urlsplit(right)
@@ -200,7 +262,7 @@ def _svg(body):
                 raster = base64.b64decode(match[2], validate=True)
             except ValueError as exc:
                 raise IconError('invalid embedded image') from exc
-            raster, _mime = normalize(raster, match[1])
+            raster, _mime = _normalize(raster, match[1])
             embedded = 'data:image/png;base64,' + base64.b64encode(raster).decode()
         out = xml.Element('{' + ns + '}' + name)
         for key, value in node.attrib.items():
@@ -243,16 +305,29 @@ def _svg(body):
 
 
 def normalize(body, mime):
+    if not isinstance(body, bytes) or not 0 < len(body) <= MAX_BYTES or mime not in MIMES:
+        raise IconError('invalid icon type or size')
+    # Cache GETs, including legacy data, use the same bounded isolation even
+    # when the Monitor itself has been configured to run without root.
+    if not DECODERS.acquire(timeout=WORKER_TIMEOUT):
+        raise IconError('local icon decoder busy')
+    try:
+        return _decode_worker(body, mime)
+    finally:
+        DECODERS.release()
+
+
+def _normalize(body, mime):
+    if os.geteuid() == 0:
+        raise IconError('image decoding as root is forbidden')
     if not isinstance(body, bytes) or not 0 < len(body) <= MAX_BYTES or mime not in MIMES:
         raise IconError('invalid icon type or size')
     if mime == 'image/svg+xml':
         return _svg(body)
     from PIL import Image
-    expected = {'image/png': 'PNG', 'image/jpeg': 'JPEG', 'image/gif': 'GIF',
-                'image/webp': 'WEBP', 'image/x-icon': 'ICO', 'image/vnd.microsoft.icon': 'ICO'}
     try:
-        with Image.open(io.BytesIO(body)) as picture:
-            if picture.format != expected[mime]:
+        with Image.open(io.BytesIO(body), formats=[RASTER_FORMATS[mime]]) as picture:
+            if picture.format != RASTER_FORMATS[mime]:
                 raise IconError('image format does not match content type')
             w, h = picture.size
             if not 0 < w <= MAX_DIMENSION or not 0 < h <= MAX_DIMENSION or w * h > MAX_PIXELS:
@@ -278,6 +353,8 @@ def _fetch(origin):
             or parsed.username is not None or parsed.password is not None or parsed.path
             or parsed.query or parsed.fragment or not parsed.port):
         raise IconError('invalid local origin')
+    if not _owns_address(parsed.hostname):
+        raise IconError('selected address does not belong to this container')
     with _open(origin + '/', page=True) as response:
         final = response.geturl()
         if not _same_origin(origin, final) or response.headers.get_content_type() not in ('text/html', 'application/xhtml+xml'):
@@ -293,22 +370,72 @@ def _fetch(origin):
     with _open(icon_url) as response:
         if response.geturl() != icon_url:
             raise IconError('icon redirect not permitted')
-        return normalize(_read(response), response.headers.get_content_type())
+        return _normalize(_read(response), response.headers.get_content_type())
 
 
-def _worker(origin):
+def _worker_command(arguments, network_fd):
+    # No guest binaries, rootfs, environment, user namespace or mount namespace.
+    # setpriv execs the host Python only after dropping credentials/capabilities.
+    command = ['/usr/bin/setpriv', '--no-new-privs']
+    if os.geteuid() == 0:
+        command += ['--reuid=65534', '--regid=65534', '--clear-groups',
+                    '--bounding-set=-all', '--inh-caps=-all', '--ambient-caps=-all']
+    command += ['--', sys.executable, '-B', str(Path(__file__).resolve()), *arguments]
+    if network_fd is not None:
+        return ['/usr/bin/nsenter', f'--net=/proc/self/fd/{network_fd}', '--no-fork', '--', *command]
+    if os.geteuid() == 0:
+        # Legacy/cache decoding cannot make any network request, even if a
+        # decoder is compromised. No CT probe is needed for a cache-only GET.
+        return ['/usr/bin/unshare', '--net', '--', *command]
+    return command
+
+
+def _run_worker(arguments, *, network_fd=None, data=None):
+    process = None
     try:
-        result = subprocess.run([sys.executable, str(Path(__file__).resolve()), '--fetch', origin],
-            capture_output=True, timeout=WORKER_TIMEOUT)
-        if result.returncode or len(result.stdout) > MAX_BYTES * 2:
-            raise IconError('local icon unavailable')
-        data = json.loads(result.stdout)
-        body = base64.b64decode(data['body'], validate=True)
-        if not body or len(body) > MAX_BYTES or data['mime'] not in ('image/png', 'image/svg+xml'):
-            raise IconError('invalid image worker response')
-        return body, data['mime']
+        with tempfile.TemporaryFile() as output:
+            process = subprocess.Popen(_worker_command(arguments, network_fd),
+                stdin=subprocess.PIPE if data is not None else subprocess.DEVNULL,
+                stdout=output, stderr=subprocess.DEVNULL, cwd='/', start_new_session=True,
+                pass_fds=() if network_fd is None else (network_fd,))
+            process.communicate(input=data, timeout=WORKER_TIMEOUT)
+            output.seek(0)
+            raw = output.read(MAX_BYTES * 2 + 1)
+            if process.returncode or len(raw) > MAX_BYTES * 2:
+                raise IconError('local icon unavailable')
+            response = json.loads(raw)
+            body = base64.b64decode(response['body'], validate=True)
+            if not body or len(body) > MAX_BYTES or response['mime'] not in ('image/png', 'image/svg+xml'):
+                raise IconError('invalid image worker response')
+            return body, response['mime']
     except (OSError, subprocess.SubprocessError, ValueError, KeyError, TypeError) as exc:
         raise IconError('local icon unavailable') from exc
+    finally:
+        if process is not None:
+            # The nsenter/setpriv chain execs without forking; also terminate
+            # any descendants a faulty decoder could have left behind.
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+            process.wait()
+
+
+def _worker(origin, network_fd):
+    return _run_worker(['--fetch', origin, str(network_fd)], network_fd=network_fd)
+
+
+def _decode_worker(body, mime):
+    data = json.dumps({'body': base64.b64encode(body).decode(), 'mime': mime}).encode()
+    return _run_worker(['--decode'], data=data)
+
+
+def _assert_worker():
+    fields = dict(line.split(':', 1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line)
+    if (os.geteuid() == 0 or os.getegid() == 0 or 0 in os.getgroups()
+            or fields.get('NoNewPrivs', '').strip() != '1'
+            or any(int(fields.get(key, '1').strip(), 16) for key in ('CapEff', 'CapPrm', 'CapAmb'))):
+        raise IconError('image worker is not isolated')
 
 
 @contextlib.contextmanager
@@ -408,11 +535,11 @@ def discover(vmid, payload):
         raise IconError('local icon lookup busy')
     try:
         origin = _origin(vmid, payload)
-        body, mime = _worker(origin)
-        # Confirm that the guest still owns the chosen address before caching.
-        if _origin(vmid, payload) != origin:
-            raise IconError('container address changed')
-        return {'logo_url': store(vmid, body, mime)}
+        with _network(vmid) as (network_fd, identity):
+            body, mime = _worker(origin, network_fd)
+            if _origin(vmid, payload) != origin or not _network_unchanged(vmid, identity):
+                raise IconError('container address or network changed')
+            return {'logo_url': store(vmid, body, mime)}
     except OSError as exc:
         raise IconError('could not cache the icon') from exc
     finally:
@@ -449,10 +576,23 @@ if __name__ == '__main__':
     import resource
     resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024,) * 2)
     resource.setrlimit(resource.RLIMIT_CPU, (4, 4))
+    resource.setrlimit(resource.RLIMIT_FSIZE, (MAX_BYTES * 2, MAX_BYTES * 2))
     try:
-        if len(sys.argv) != 3 or sys.argv[1] != '--fetch':
+        _assert_worker()
+        if len(sys.argv) == 4 and sys.argv[1] == '--fetch':
+            fd = int(sys.argv[3])
+            if _ns_id(os.fstat(fd)) != _ns_id(Path('/proc/self/ns/net').stat()):
+                raise IconError('image worker network mismatch')
+            os.close(fd)
+            body, mime = _fetch(sys.argv[2])
+        elif len(sys.argv) == 2 and sys.argv[1] == '--decode':
+            raw = sys.stdin.buffer.read(MAX_BYTES * 2 + 1)
+            if len(raw) > MAX_BYTES * 2:
+                raise IconError('image input too large')
+            data = json.loads(raw)
+            body, mime = _normalize(base64.b64decode(data['body'], validate=True), data['mime'])
+        else:
             raise IconError('invalid worker arguments')
-        body, mime = _fetch(sys.argv[2])
         print(json.dumps({'mime': mime, 'body': base64.b64encode(body).decode()}))
     except Exception:
         sys.exit(1)

AppImage/scripts/tests/test_local_app_icon_isolation.py

@@ -0,0 +1,220 @@
+"""Security boundaries independent of UI/cache successes; no live CT required."""
+import base64
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import local_app_icons as icons
+
+
+def png():
+    output = io.BytesIO()
+    Image.new('RGB', (8, 8)).save(output, 'PNG')
+    return output.getvalue()
+
+
+class Isolation(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+
+    def test_root_normalization_never_opens_an_image_in_the_monitor(self):
+        data = png()
+        with patch.object(icons.os, 'geteuid', return_value=0), \
+                patch.object(icons, '_decode_worker', return_value=(data, 'image/png'), create=True) as child, \
+                patch.object(Image, 'open', side_effect=AssertionError('root decoder forbidden')) as opened:
+            self.assertEqual(icons.normalize(data, 'image/png'), (data, 'image/png'))
+        child.assert_called_once_with(data, 'image/png')
+        opened.assert_not_called()
+
+    def test_only_the_selected_raster_decoder_is_enabled(self):
+        data = png()
+        with patch.object(icons.os, 'geteuid', return_value=65534), patch.object(Image, 'open', wraps=Image.open) as opened:
+            getattr(icons, '_normalize', icons.normalize)(data, 'image/png')
+        self.assertEqual(opened.call_args.kwargs.get('formats'), ['PNG'])
+
+    def test_unprivileged_monitor_also_uses_a_bounded_decoder(self):
+        data=png()
+        with patch.object(icons.os,'geteuid',return_value=1001), \
+                patch.object(icons,'_decode_worker',return_value=(data,'image/png')) as decoder:
+            self.assertEqual(icons.normalize(data,'image/png'),(data,'image/png'))
+        decoder.assert_called_once_with(data,'image/png')
+
+    def test_direct_root_decoder_is_rejected_before_pillow(self):
+        data=png()
+        with patch.object(icons.os,'geteuid',return_value=0),patch.object(Image,'open') as opened:
+            with self.assertRaises(icons.IconError):icons._normalize(data,'image/png')
+        opened.assert_not_called()
+
+    def test_saturated_decoder_has_zero_child_or_image_calls(self):
+        data=png();slots=Mock();slots.acquire.return_value=False
+        with patch.object(icons,'DECODERS',slots),patch.object(icons,'_decode_worker') as worker,patch.object(Image,'open') as opened:
+            with self.assertRaises(icons.IconError):icons.normalize(data,'image/png')
+        worker.assert_not_called();opened.assert_not_called();slots.release.assert_not_called()
+
+    def test_decoder_failure_always_releases_its_slot(self):
+        data=png();slots=Mock();slots.acquire.return_value=True
+        with patch.object(icons,'DECODERS',slots),patch.object(icons,'_decode_worker',side_effect=icons.IconError('unavailable')):
+            with self.assertRaises(icons.IconError):icons.normalize(data,'image/png')
+        slots.release.assert_called_once()
+
+    def test_foreign_address_has_zero_http_calls_even_when_lxc_reports_it(self):
+        with patch.object(icons, '_owns_address', return_value=False, create=True), \
+                patch.object(icons, '_open', side_effect=AssertionError('foreign network forbidden')) as http:
+            with self.assertRaises(icons.IconError):
+                icons._fetch('http://10.0.0.99:8080')
+        http.assert_not_called()
+
+    def fake_proc(self):
+        for folder in ('self/ns', '2468/ns'):
+            (self.root / folder).mkdir(parents=True)
+            (self.root / folder / 'net').write_text('namespace fixture')
+        (self.root / '2468/stat').write_text('2468 (container (init)) S ' + '0 ' * 18 + '12345 0\n')
+
+    def test_namespace_descriptor_is_pinned_and_closed(self):
+        self.fake_proc()
+        with patch.object(icons, 'PROC', self.root), patch.object(icons, '_run', return_value='2468\n') as run:
+            with icons._network(101) as (fd, identity):
+                self.assertEqual(identity[0:2], (2468, '12345'))
+                self.assertEqual(os.fstat(fd).st_ino, (self.root / '2468/ns/net').stat().st_ino)
+                self.assertTrue(icons._network_unchanged(101, identity))
+            with self.assertRaises(OSError):os.fstat(fd)
+        self.assertTrue(all(c.args[0] == ['lxc-info', '-n', '101', '-pH'] for c in run.call_args_list))
+
+    def test_host_namespace_pid_reuse_and_changed_pid_never_fetch_or_store(self):
+        for case in ('host', 'pid', 'start', 'namespace'):
+            with self.subTest(case=case), tempfile.TemporaryDirectory() as temporary:
+                self.root = Path(temporary);self.fake_proc()
+                def identify(args):
+                    if case == 'pid':return '2469' if identify.calls else self.first(identify)
+                    if identify.calls:
+                        if case == 'start':(self.root / '2468/stat').write_text('2468 (init) S ' + '0 ' * 18 + '54321\n')
+                        if case == 'namespace':
+                            (self.root / '2468/ns/new').write_text('replacement')
+                            os.replace(self.root / '2468/ns/new', self.root / '2468/ns/net')
+                    identify.calls += 1
+                    return '2468'
+                identify.calls=0
+                if case=='host':
+                    (self.root / '2468/ns/net').unlink()
+                    os.link(self.root / 'self/ns/net', self.root / '2468/ns/net')
+                with patch.object(icons, 'PROC', self.root), patch.object(icons, '_run', side_effect=identify), \
+                        patch.object(icons, '_origin', return_value='http://10.0.0.42:80'), \
+                        patch.object(icons, '_worker') as worker, patch.object(icons, 'store') as store:
+                    with self.assertRaises(icons.IconError):icons.discover(101, {'port':80,'scheme':'http'})
+                worker.assert_not_called();store.assert_not_called()
+
+    @staticmethod
+    def first(function):
+        function.calls+=1
+        return '2468'
+
+    def test_missing_or_invalid_pid_cannot_start_a_worker(self):
+        for value in ('', '0', '1', '-3', '2468\n2469', 'text'):
+            with self.subTest(pid=value), patch.object(icons, '_run', return_value=value), \
+                    patch.object(icons, '_origin', return_value='http://10.0.0.42:80'), \
+                    patch.object(icons, '_worker') as worker, patch.object(icons, 'store') as store, \
+                    patch.object(icons.os, 'open') as opened:
+                with self.assertRaises(icons.IconError):icons.discover(101, {'port':80,'scheme':'http'})
+                worker.assert_not_called();store.assert_not_called();opened.assert_not_called()
+
+    def test_partial_process_identity_and_missing_namespace_fail_without_effects(self):
+        for value in ('', '2468 (truncated)', '2468 (init) S 0 0'):
+            with self.subTest(stat=value),tempfile.TemporaryDirectory() as temporary:
+                root=Path(temporary);(root/'2468').mkdir();(root/'2468/stat').write_text(value)
+                with patch.object(icons,'PROC',root),patch.object(icons,'_run',return_value='2468'), \
+                        patch.object(icons,'_origin',return_value='http://10.0.0.42:80'), \
+                        patch.object(icons,'_worker') as worker,patch.object(icons,'store') as store:
+                    with self.assertRaises(icons.IconError):icons.discover(101,{'port':80,'scheme':'http'})
+                worker.assert_not_called();store.assert_not_called()
+        self.fake_proc();(self.root/'2468/ns/net').unlink()
+        with patch.object(icons,'PROC',self.root),patch.object(icons,'_run',return_value='2468'), \
+                patch.object(icons,'_origin',return_value='http://10.0.0.42:80'), \
+                patch.object(icons,'_worker') as worker,patch.object(icons,'store') as store:
+            with self.assertRaises(icons.IconError):icons.discover(101,{'port':80,'scheme':'http'})
+        worker.assert_not_called();store.assert_not_called()
+
+    def test_failed_init_pid_command_has_zero_network_or_cache_effects(self):
+        with patch.object(icons,'_run',side_effect=icons.IconError('lxc-info timed out')), \
+                patch.object(icons,'_origin',return_value='http://10.0.0.42:80'), \
+                patch.object(icons,'_worker') as worker,patch.object(icons,'store') as store:
+            with self.assertRaises(icons.IconError):icons.discover(101,{'port':80,'scheme':'http'})
+        worker.assert_not_called();store.assert_not_called()
+
+    def test_guest_restart_after_worker_preserves_existing_cache(self):
+        with patch.object(icons, '_origin', return_value='http://10.0.0.42:80'), \
+                patch.object(icons, '_network', return_value=contextlib.nullcontext((81,'old'))), \
+                patch.object(icons, '_worker', return_value=(png(),'image/png')), \
+                patch.object(icons, '_network_unchanged', return_value=False), patch.object(icons,'store') as store:
+            with self.assertRaises(icons.IconError):icons.discover(101, {'port':80,'scheme':'http'})
+        store.assert_not_called()
+
+    def test_only_ct_network_namespace_and_unprivileged_credentials_are_selected(self):
+        with patch.object(icons.os,'geteuid',return_value=0):
+            command=icons._worker_command(['--fetch','http://10.0.0.42:80','81'],81)
+            decoder=icons._worker_command(['--decode'],None)
+        self.assertEqual(command[:4],['/usr/bin/nsenter','--net=/proc/self/fd/81','--no-fork','--'])
+        for required in ('--reuid=65534','--regid=65534','--clear-groups','--no-new-privs','--bounding-set=-all','--inh-caps=-all','--ambient-caps=-all'):
+            self.assertIn(required,command);self.assertIn(required,decoder)
+        self.assertEqual(decoder[:3],['/usr/bin/unshare','--net','--'])
+        self.assertNotIn('--all',command);self.assertNotIn('--mount',command);self.assertNotIn('--user',command)
+
+    def test_failure_to_enter_namespace_has_no_host_fallback(self):
+        with patch.object(icons.subprocess,'Popen',side_effect=OSError('nsenter unavailable')) as start:
+            with self.assertRaises(icons.IconError):icons._worker('http://10.0.0.42:80',81)
+        self.assertEqual(start.call_count,1)
+        self.assertEqual(start.call_args.args[0][0],'/usr/bin/nsenter')
+        self.assertEqual(start.call_args.kwargs['pass_fds'],(81,))
+
+    def test_address_ownership_uses_only_namespace_local_interfaces(self):
+        for address in ('10.0.0.42','fd12::42'):
+            with patch.object(icons,'_run',return_value=json.dumps([{'addr_info':[{'local':address}]}])) as run:
+                self.assertTrue(icons._owns_address(address));self.assertFalse(icons._owns_address('10.0.0.99'))
+                self.assertEqual(run.call_args.args[0],['/usr/sbin/ip','-json','address','show'])
+        with patch.object(icons,'_run',return_value='partial JSON'), patch.object(icons,'_open') as network:
+            with self.assertRaises(icons.IconError):icons._fetch('http://10.0.0.42:80')
+            network.assert_not_called()
+
+    def test_worker_security_guard_precedes_decode_or_network(self):
+        fields='NoNewPrivs:\t1\nCapEff:\t0\nCapPrm:\t0\nCapAmb:\t0\n'
+        for uid,gid,groups,body in ((0,65534,[],fields),(65534,0,[],fields),
+                (65534,65534,[0],fields),(65534,65534,[],fields.replace('NoNewPrivs:\t1','NoNewPrivs:\t0')),
+                *[(65534,65534,[],fields.replace(key+':\t0',key+':\t1')) for key in ('CapEff','CapPrm','CapAmb')]):
+            with patch.object(icons.os,'geteuid',return_value=uid),patch.object(icons.os,'getegid',return_value=gid), \
+                    patch.object(icons.os,'getgroups',return_value=groups),patch.object(Path,'read_text',return_value=body):
+                with self.assertRaises(icons.IconError):icons._assert_worker()
+
+    def test_oversized_and_malformed_worker_output_is_rejected(self):
+        helper=self.root/'output.py'
+        for body in ('x'*(icons.MAX_BYTES*2+1),'not-json',json.dumps({'mime':'image/png','body':'bad base64'})):
+            helper.write_text('import sys\nsys.stdout.write('+repr(body)+')\n')
+            with patch.object(icons,'_worker_command',return_value=[sys.executable,str(helper)]):
+                with self.assertRaises(icons.IconError):icons._run_worker(['--decode'])
+
+    def test_root_legacy_and_current_cache_decode_without_guest_probes(self):
+        data=png();root=self.root/'101';root.mkdir()
+        for legacy in (False,True):
+            token=('a' if legacy else 'b')*32
+            if legacy:
+                (root/(token+'.bin')).write_bytes(data)
+                (root/(token+'.json')).write_text(json.dumps({'content_type':'image/png'}))
+            else:(root/(token+'.icon')).write_text(json.dumps({'mime':'image/png','body':base64.b64encode(data).decode()}))
+            with patch.object(icons,'CACHE',self.root),patch.object(icons.os,'geteuid',return_value=0), \
+                    patch.object(icons,'_decode_worker',return_value=(data,'image/png')) as decoder, \
+                    patch.object(icons,'_run') as guest,patch.object(Image,'open') as raster:
+                self.assertEqual(icons.read(101,token),(data,'image/png'))
+                decoder.assert_called_once();guest.assert_not_called();raster.assert_not_called()
+
+
+if __name__=='__main__':unittest.main()

AppImage/scripts/tests/test_local_app_icons.py

@@ -1,6 +1,7 @@
 """Offline icon contracts: actual codecs, HTTP, cache transactions and routes."""
 import ast
 import base64
+import contextlib
 from concurrent.futures import ThreadPoolExecutor
 from functools import wraps
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -49,6 +50,12 @@ class Icons(unittest.TestCase):
             self.addCleanup(p.stop)
         self.png = picture()
         self.payload = {'port': 8080, 'scheme': 'http'}
+        # Namespace identity is exercised separately by test_local_app_icon_isolation.
+        for name, value in (('_network', contextlib.nullcontext((81, 'fixture'))),
+                            ('_network_unchanged', True)):
+            p = patch.object(icons, name, return_value=value)
+            p.start()
+            self.addCleanup(p.stop)
 
     def resolve(self, args):
         if args[:2] == ['/usr/sbin/pct', 'status']:
@@ -137,7 +144,7 @@ class Icons(unittest.TestCase):
     def test_explicit_lookup_derives_address_rechecks_and_never_writes_apps(self):
         with patch.object(icons, '_run', side_effect=self.resolve) as run, patch.object(icons, '_worker', return_value=(self.png, 'image/png')) as worker:
             result = icons.discover(101, self.payload)
-        worker.assert_called_once_with('http://10.0.0.42:8080')
+        worker.assert_called_once_with('http://10.0.0.42:8080', 81)
         self.assertEqual(run.call_count, 6)
         self.assertFalse((self.apps / '101.json').exists())
         self.assertEqual(icons.read(101, result['logo_url'].split('/')[-1])[1], 'image/png')
@@ -191,7 +198,7 @@ class Icons(unittest.TestCase):
                 return '\n'.join(addresses + addresses) if args[0] == 'lxc-info' else self.resolve(args)
             with patch.object(icons, '_run', side_effect=resolve), patch.object(icons, '_worker', return_value=(self.png, 'image/png')) as worker:
                 icons.discover(101, self.payload)
-                worker.assert_called_once_with(f'http://{expected}:8080')
+                worker.assert_called_once_with(f'http://{expected}:8080', 81)
 
     def test_address_change_or_stop_during_request_cannot_cache(self):
         for second in ['http://10.0.0.43:8080', icons.IconError('stopped')]:
@@ -207,17 +214,17 @@ class Icons(unittest.TestCase):
             with self.assertRaises(icons.IconError):
                 icons.discover(101, self.payload)
             origin.assert_not_called()
-        with patch.object(icons.subprocess, 'run', side_effect=subprocess.TimeoutExpired('worker', 8)):
+        with patch.object(icons.subprocess, 'Popen', side_effect=OSError('cannot start worker')):
             with self.assertRaises(icons.IconError):
-                icons._worker('http://10.0.0.42:80')
+                icons._worker('http://10.0.0.42:80', 81)
 
     def test_real_worker_timeout_kills_child_and_returns_promptly(self):
         worker = self.root / 'slow.py'
         worker.write_text('import time\ntime.sleep(20)\n')
         started = time.monotonic()
-        with patch.object(icons, '__file__', str(worker)), patch.object(icons, 'WORKER_TIMEOUT', .15):
+        with patch.object(icons, '_worker_command', return_value=[sys.executable, str(worker)]), patch.object(icons, 'WORKER_TIMEOUT', .15):
             with self.assertRaises(icons.IconError):
-                icons._worker('http://10.0.0.42:80')
+                icons._run_worker(['--decode'])
         self.assertLess(time.monotonic() - started, 2)
 
     def test_proxy_environment_is_ignored_and_private_https_context_is_explicit(self):
@@ -396,7 +403,7 @@ class HttpContract(unittest.TestCase):
         try:
             origin = f'http://127.0.0.1:{server.server_port}'
             # Production rejects loopback; this fixture tests only the HTTP contract.
-            with patch.object(icons, '_private_ip', return_value=True):
+            with patch.object(icons, '_private_ip', return_value=True), patch.object(icons, '_owns_address', return_value=True):
                 for case, paths in [('relative', ['/', '/icon.png']), ('redirect', ['/', '/login/', '/login/icon.png']), ('fallback', ['/', '/favicon.ico'])]:
                     mode[0] = case; requests.clear()
                     self.assertEqual(icons._fetch(origin)[1], 'image/png')