Test our handling of the SSH agent "query" double-response behavior
Marco Ricci

Marco Ricci commited on 2026-06-13 18:52:55
Zeige 1 geänderte Dateien mit 124 Einfügungen und 1 Löschungen.


Add tests to verify that we gracefully react to an SSH agent replying
twice to a "query" extension request.
... ...
@@ -16,7 +16,7 @@ import re
16 16
 import socket
17 17
 import sys
18 18
 import types
19
-from typing import TYPE_CHECKING, NamedTuple
19
+from typing import TYPE_CHECKING, Final, NamedTuple
20 20
 
21 21
 import click
22 22
 import hypothesis
... ...
@@ -1440,6 +1440,129 @@ class TestAgentErrorResponses:
1440 1440
             ):
1441 1441
                 client.query_extensions()
1442 1442
 
1443
+    def _query_response(
1444
+        self,
1445
+        /,
1446
+        *extensions: str,
1447
+        code: Literal[
1448
+            _types.SSH_AGENT.EXTENSION_RESPONSE, _types.SSH_AGENT.SUCCESS
1449
+        ] = _types.SSH_AGENT.EXTENSION_RESPONSE,
1450
+    ) -> Response:
1451
+        """Construct a response object from the given extensions.
1452
+
1453
+        Args:
1454
+            extensions:
1455
+                Names of extensions to advertise support for.
1456
+            code:
1457
+                The response code of the message.
1458
+
1459
+        Returns:
1460
+            An appropriately constructed response object.
1461
+
1462
+        """
1463
+        payload = bytearray(b"\x00\x00\x00\x05query")
1464
+        for ext in extensions:
1465
+            payload.extend(TestStaticFunctionality.as_ssh_string(ext.encode()))
1466
+        return self.Response(code, bytes(payload))
1467
+
1468
+    IDENTITIES_REQUEST: Final = Request(
1469
+        _types.SSH_AGENTC.REQUEST_IDENTITIES, b""
1470
+    )
1471
+    IDENTITIES_RESPONSE: Final = Response(
1472
+        _types.SSH_AGENT.IDENTITIES_ANSWER, b"\x00\x00\x00\x00"
1473
+    )
1474
+    QUERY_REQUEST: Final = Request(
1475
+        _types.SSH_AGENTC.EXTENSION, b"\x00\x00\x00\x05query"
1476
+    )
1477
+    EXTRA_RESPONSE: Final = Response(_types.SSH_AGENT.SUCCESS, b"")
1478
+
1479
+    @pytest.mark.parametrize(
1480
+        ["success_after_query", "extensions"],
1481
+        [
1482
+            pytest.param(
1483
+                False,
1484
+                Response(_types.SSH_AGENT.FAILURE, b""),
1485
+                id="openssh-10.2",
1486
+            ),
1487
+            pytest.param(
1488
+                True,
1489
+                Response(_types.SSH_AGENT.FAILURE, b""),
1490
+                id="openssh-10.2-mocked",
1491
+                marks=[
1492
+                    pytest.mark.xfail(
1493
+                        reason="specific to OpenSSH 10.3 output", strict=True
1494
+                    )
1495
+                ],
1496
+            ),
1497
+            pytest.param(
1498
+                False, ["session-bind@openssh.com"], id="openssh-fixed"
1499
+            ),
1500
+            pytest.param(
1501
+                True, ["session-bind@openssh.com"], id="openssh-10.3"
1502
+            ),
1503
+            pytest.param(
1504
+                False,
1505
+                [
1506
+                    "add-ppk@putty.projects.tartarus.org",
1507
+                    "reencrypt@putty.projects.tartarus.org",
1508
+                    "reencrypt-all@putty.projects.tartarus.org",
1509
+                    "list-extended@putty.projects.tartarus.org",
1510
+                ],
1511
+                id="pageant",
1512
+            ),
1513
+            pytest.param(
1514
+                True,
1515
+                [
1516
+                    "add-ppk@putty.projects.tartarus.org",
1517
+                    "reencrypt@putty.projects.tartarus.org",
1518
+                    "reencrypt-all@putty.projects.tartarus.org",
1519
+                    "list-extended@putty.projects.tartarus.org",
1520
+                ],
1521
+                id="pageant-mocked",
1522
+                marks=[
1523
+                    pytest.mark.xfail(
1524
+                        reason="specific to OpenSSH 10.3 output", strict=True
1525
+                    )
1526
+                ],
1527
+            ),
1528
+        ],
1529
+    )
1530
+    def test_tolerating_extra_success_after_extension_response(
1531
+        self, success_after_query: bool, extensions: list[str] | Response
1532
+    ) -> None:
1533
+        """Tolerate an extra empty SUCCESS message from a "query" EXTENSION.
1534
+
1535
+        Specifically, in a "query"
1536
+        [`EXTENSION`][_types.SSH_AGENTC.EXTENSION] request, tolerate
1537
+        getting potentially both an
1538
+        [`EXTENSION_RESPONSE`][_types.SSH_AGENT.EXTENSION_RESPONSE] or
1539
+        [`SUCCESS`][_types.SSH_AGENT.SUCCESS] message and an additional
1540
+        empty [`SUCCESS`][_types.SSH_AGENT.SUCCESS] message afterwards,
1541
+        instead of only the first message.  This is spec-uncompliant
1542
+        behavior observed in `ssh-agent` from OpenSSH 10.3.  (And since
1543
+        then reported and acknowledged upstream, to be fixed in OpenSSH
1544
+        10.4: [OpenSSH Bug 3967][OPENSSH_BZ3967].)
1545
+
1546
+        [OPENSSH_BZ3967]: https://bugzilla.mindrot.org/show_bug.cgi?id=3967
1547
+
1548
+        """
1549
+        query_response = (
1550
+            extensions
1551
+            if isinstance(extensions, self.Response)
1552
+            else self._query_response(*extensions)
1553
+        )
1554
+        request_responses_map = {
1555
+            self.IDENTITIES_REQUEST: [self.IDENTITIES_RESPONSE],
1556
+            self.QUERY_REQUEST: [query_response, self.EXTRA_RESPONSE]
1557
+            if success_after_query
1558
+            else [query_response],
1559
+        }
1560
+        with self._setup_agent_and_request_handler(
1561
+            request_responses_map
1562
+        ) as contexts:
1563
+            _, client = contexts
1564
+            client.query_extensions()
1565
+
1443 1566
 
1444 1567
 class TestSSHAgentSocketProviderFailures:
1445 1568
     """Test SSH agent socket provider-specific failures."""
1446 1569