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
near-zero risk of an injection vulnerability.  Therefore, use a separate
unrestricted agent connection for these commands.  Also use a separate
function to read agent hello messages sent upon connection.

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({})
+        self.client_write(await self.read_hello(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
@@ -742,11 +754,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:
@@ -821,7 +835,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()
@@ -1018,43 +1033,68 @@ 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)
+            await self.read_hello(reader)
         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
+    async def read_hello(self, agent_reader: asyncio.StreamReader) -> bytes:
+        while True:
+            line = await agent_reader.readline()
+            if not line.endswith(b'\n'):
+                raise ProtocolError("premature EOF from agent connection")
+            if b'\n' in line[:-1]:
+                raise ProtocolError("newline in readline() result???")
+            if line.startswith(b'#'):
+                continue
+            if line == b'OK' or line.startswith(b'OK '):
+                return line
+            raise ProtocolError("agent responded with something other than 'OK'"
+                                " to initial connection")
+
+    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)
         if untrusted_line.startswith(b'#'):
@@ -1295,7 +1335,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
 
@@ -1393,7 +1435,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