Use unrestricted connection for HAVEKEY --list and KEYINFO --list

These commands are forbidden over a restricted connection to the agent,
but GnuPG wars if they are not present and Sequoia Chameleon requires
them.  Fortunately, they are trivial to sanitize input for, so there is
zero risk of an injection vulnerability.  Therefore, use a separate
unrestricted agent connection for these commands.

Author: DemiMarie

splitgpg2/__init__.py

@@ -218,6 +218,7 @@ class GpgServer:
     commands: Dict[bytes, 'NoneCallback']
     seen_data: bool
     config_loaded: bool
+    agent_unrestricted_socket_path: Optional[str]
     agent_socket_path: Optional[str]
     agent_reader: Optional[asyncio.StreamReader]
     agent_writer: Optional[asyncio.StreamWriter]
@@ -245,6 +246,7 @@ class GpgServer:
                  'seen_data',
                  'config_loaded',
                  'agent_socket_path',
+                 'agent_unrestricted_socket_path',
                  'agent_reader',
                  'agent_writer',
                  'source_keyring_dir',
@@ -275,6 +277,7 @@ def __init__(self, reader: asyncio.StreamReader,
 
         self.log = logging.getLogger('splitgpg2.Server')
         self.agent_socket_path = None
+        self.agent_unrestricted_socket_path = None
         self.agent_reader: Optional[asyncio.StreamReader] = None
         self.agent_writer: Optional[asyncio.StreamWriter] = None
 
@@ -463,11 +466,20 @@ async def connect_agent(self) -> None:
         # a message.
         # The filtering done by split-gpg2 is far stronger than anything the agent does
         # internally.
-        socket_field = b'agent-socket:'
+        unrestricted_socket_field = b'agent-socket'
+        socket_field = unrestricted_socket_field if self.allow_keygen else b'agent-extra-socket'
         # search for agent-socket:/run/user/1000/gnupg/S.gpg-agent
-        agent_socket_path = [d.split(b':', 1)[1] for d in dirs.splitlines()
-                             if d.startswith(socket_field)][0]
-        self.agent_socket_path = agent_socket_path.decode()
+        for d in dirs.splitlines():
+            key, value = d.split(b':')
+            if key == socket_field:
+                self.agent_socket_path = value.decode("UTF-8", "surrogateescape")
+            if key == unrestricted_socket_field:
+                self.agent_unrestricted_socket_path = value.decode("UTF-8", "surrogateescape")
+            if ((self.agent_unrestricted_socket_path is not None) and
+                (self.agent_socket_path is not None)):
+                break
+        else:
+            raise RuntimeError("bad output from gpgconf")
 
         self.agent_reader, self.agent_writer = await asyncio.open_unix_connection(
                 path=self.agent_socket_path)
@@ -476,7 +488,7 @@ async def connect_agent(self) -> None:
             self.notify('connected')
 
         # wait for agent hello
-        await self.handle_agent_response({})
+        await self.handle_agent_response({}, self.agent_reader)
 
     def close(self, reason: str, log_level: int = logging.ERROR) -> None:
         self.log.log(log_level, '%s; Closing!', reason)
@@ -563,8 +575,8 @@ def default_options() -> Dict[bytes, Tuple[OptionHandlingType, Optional[bytes]]]
             b'lc-messages': (OptionHandlingType.fake, b'OK'),
             b'putenv': (OptionHandlingType.fake, b'OK'),
             b'pinentry-mode': (OptionHandlingType.fake, b'ERR 67108924 Not supported <GPG Agent>'),
-            b'allow-pinentry-notify': (OptionHandlingType.verify, None),
-            b'agent-awareness': (OptionHandlingType.verify, b'2.1.0')
+            b'allow-pinentry-notify': (OptionHandlingType.fake, b'OK'),
+            b'agent-awareness': (OptionHandlingType.verify, b'2.1.0'),
         }
 
     @staticmethod
@@ -656,7 +668,9 @@ def verify_keygrip_arguments(min_count: int, max_count: int,
                 pass
             elif untrusted_args[6] == 61: # ASCII '='
                 # 1000 is the default value used by gpg2
-                sanitize_int(untrusted_args[7:], 1, 1000)
+                if untrusted_args != b'--list=%d' % sanitize_int(untrusted_args[7:], 1, 1000):
+                    assert False, "bug in filtering"
+                    raise Filtered
             else:
                 raise Filtered
         else:
@@ -744,11 +758,13 @@ async def command_HAVEKEY(self, untrusted_args: Optional[bytes]) -> None:
             raise Filtered
         # upper keygrip limit is arbitary
         args = self.verify_keygrip_arguments(1, 200, untrusted_args, True)
-        await self.send_agent_command(b'HAVEKEY', args)
+        unrestricted = args.startswith(b'--list') and not self.allow_keygen
+        await self.send_agent_command(b'HAVEKEY', args, unrestricted)
 
     async def command_KEYINFO(self, untrusted_args: Optional[bytes]) -> None:
         args = self.verify_keygrip_arguments(1, 1, untrusted_args, True)
-        await self.send_agent_command(b'KEYINFO', args)
+        unrestricted = args.startswith(b'--list') and not self.allow_keygen
+        await self.send_agent_command(b'KEYINFO', args, unrestricted)
 
     async def command_GENKEY(self, untrusted_args: Optional[bytes]) -> None:
         if not self.allow_keygen:
@@ -823,7 +839,8 @@ async def setkeydesc(self, keygrip: bytes) -> None:
                     key.fingerprint,
                     subkey_desc)
 
-        self.agent_write(b'SETKEYDESC %s\n' % self.percent_plus_escape(desc))
+        assert self.agent_writer is not None, "no writer?"
+        self.agent_write(b'SETKEYDESC %s\n' % self.percent_plus_escape(desc), self.agent_writer)
 
         assert self.agent_reader is not None
         untrusted_line = await self.agent_reader.readline()
@@ -1020,43 +1037,56 @@ def get_inquires_for_command(self, command: bytes) -> Dict[bytes, 'ArgCallback']
             }
         return {}
 
-    async def send_agent_command(self, command: bytes, args: Optional[bytes]) -> None:
+    async def send_agent_command(self, command: bytes, args: Optional[bytes],
+                                 unrestricted: bool=False) -> None:
         """ Sends command to local gpg agent and handle the response """
         expected_inquires = self.get_inquires_for_command(command)
-        if args:
-            if not self.command_argument_regex.match(args):
-                raise AssertionError("BUG: corrupt command about to be sent to agent!")
-            cmd_with_args = command + b' ' + args + b'\n'
+        assert self.agent_reader is not None, "no reader?"
+        assert self.agent_writer is not None, "no writer?"
+        if unrestricted and not self.allow_keygen:
+            reader, writer = await asyncio.open_unix_connection(
+                    self.agent_unrestricted_socket_path)
+            line = await reader.readline()
+            if not line.startswith(b'OK '):
+                raise Filtered # TODO: better handle error
         else:
-            cmd_with_args = command + b'\n'
-        self.agent_write(cmd_with_args)
-        while True:
-            more_expected = await self.handle_agent_response(
-                    expected_inquires=expected_inquires)
-            if not more_expected:
-                break
+            reader, writer = self.agent_reader, self.agent_writer
+        try:
+            if args:
+                if not self.command_argument_regex.match(args):
+                    raise AssertionError("BUG: corrupt command about to be sent to agent!")
+                cmd_with_args = command + b' ' + args + b'\n'
+            else:
+                cmd_with_args = command + b'\n'
+            self.agent_write(cmd_with_args, writer)
+            while True:
+                more_expected = await self.handle_agent_response(expected_inquires, reader)
+                if not more_expected:
+                    break
+        finally:
+            if reader is not self.agent_reader:
+                writer.close()
 
-    def agent_write(self, data: bytes) -> None:
-        writer = self.agent_writer
+    def agent_write(self, data: bytes, writer: asyncio.StreamWriter) -> None:
         assert writer is not None, 'agent_write called with no agent writer?'
         self.log_io('A <<<', data)
         writer.write(data)
 
     async def handle_agent_response(self,
-                                    expected_inquires: Dict[bytes, 'ArgCallback']) -> bool:
+                                    expected_inquires: Dict[bytes, 'ArgCallback'],
+                                    agent_reader: asyncio.StreamReader) -> bool:
         """ Receive and handle one agent response. Return whether there are
         more expected """
-        assert self.agent_reader is not None
         assert self.client_writer is not None
         if self.client_writer.is_closing():
             # If something went wrong, agent might send back junk.
             # Discard all remaining data from agent and return.
-            while await self.agent_reader.read(1024):
+            while await agent_reader.read(1024):
                 pass
             return False
         # We generally consider the agent as trusted. But since the client can
         # determine part of the response we handle this here as untrusted.
-        untrusted_line = await self.agent_reader.readline()
+        untrusted_line = await agent_reader.readline()
         untrusted_line = untrusted_line.rstrip(b'\n')
         self.log_io('A >>>', untrusted_line)
         untrusted_res, untrusted_args = extract_args(untrusted_line)
@@ -1294,7 +1324,9 @@ async def inquire_command_D(self, validate_sexp: 'SExprValidator', *,
             raise Filtered from e
         args = untrusted_sexp
 
-        self.agent_write(b'D ' + self.escape_D(self.serialize_sexpr(args)) + b'\n')
+        assert self.agent_writer is not None, "no writer?"
+        self.agent_write(b'D ' + self.escape_D(self.serialize_sexpr(args)) + b'\n',
+                         self.agent_writer)
         self.seen_data = True
         return True
 
@@ -1392,7 +1424,8 @@ def serialize_item(item: 'SExpr') -> bytes:
     async def inquire_command_END(self, *, untrusted_args: bytes) -> bool:
         if untrusted_args:
             raise Filtered('unexpected arguments to END')
-        self.agent_write(b'END\n')
+        assert self.agent_writer is not None, "no writer?"
+        self.agent_write(b'END\n', self.agent_writer)
         return False
 
     # endregion