Turn the SSH agent protocol message numbers into ints
Marco Ricci

Marco Ricci commited on 2026-06-13 15:56:18
Zeige 6 geänderte Dateien mit 23 Einfügungen und 36 Löschungen.


The `SSH_AGENT` and `SSH_AGENTC` types now inherit from `int`, because
SSH agent protocol message numbers are 8-bit unsigned integers.  This
cuts down on a lot of (previously necessary) type conversion steps
whenever mixed groups of message numbers were involved (e.g. the
`SSHAgentClient.request` `response_code` argument), or when message
numbers needed to be compared with each other (e.g. tests).
... ...
@@ -651,7 +651,7 @@ class SSHKeyCommentPair(NamedTuple, Generic[T_Buffer]):
651 651
         )
652 652
 
653 653
 
654
-class SSH_AGENTC(enum.Enum):  # noqa: N801
654
+class SSH_AGENTC(int, enum.Enum):  # noqa: N801
655 655
     """SSH agent protocol numbers: client requests.
656 656
 
657 657
     Attributes:
... ...
@@ -688,7 +688,7 @@ class SSH_AGENTC(enum.Enum):  # noqa: N801
688 688
     """"""
689 689
 
690 690
 
691
-class SSH_AGENT(enum.Enum):  # noqa: N801
691
+class SSH_AGENT(int, enum.Enum):  # noqa: N801
692 692
     """SSH agent protocol numbers: server replies.
693 693
 
694 694
     Attributes:
... ...
@@ -66,7 +66,7 @@ class SSHAgentFailedError(RuntimeError):
66 66
         # TODO(the-13th-letter): Rewrite using structural pattern matching.
67 67
         # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
68 68
         if self.args == (  # pragma: no branch
69
-            _types.SSH_AGENT.FAILURE.value,
69
+            _types.SSH_AGENT.FAILURE,
70 70
             b"",
71 71
         ):
72 72
             return "The SSH agent failed to complete the request"
... ...
@@ -539,14 +539,8 @@ class SSHAgentClient:
539 539
                 any of them.
540 540
 
541 541
         """
542
-        if isinstance(  # pragma: no branch
543
-            response_code, (int, _types.SSH_AGENT)
544
-        ):
542
+        if isinstance(response_code, int):  # pragma: no branch
545 543
             response_code = frozenset({response_code})
546
-        if response_code is not None:  # pragma: no branch
547
-            response_code = frozenset({
548
-                c if isinstance(c, int) else c.value for c in response_code
549
-            })
550 544
         payload = memoryview(payload)
551 545
         request_message = bytearray([
552 546
             code if isinstance(code, int) else code.value
... ...
@@ -679,7 +673,7 @@ class SSHAgentClient:
679 673
         return bytes(
680 674
             self.unstring(
681 675
                 self.request(
682
-                    _types.SSH_AGENTC.SIGN_REQUEST.value,
676
+                    _types.SSH_AGENTC.SIGN_REQUEST,
683 677
                     request_data,
684 678
                     response_code=_types.SSH_AGENT.SIGN_RESPONSE,
685 679
                 )
... ...
@@ -384,7 +384,7 @@ class StubbedSSHAgentSocket:
384 384
         try:
385 385
             self.parse_client_request_and_dispatch()
386 386
         except ValueError:
387
-            payload = int.to_bytes(_types.SSH_AGENT.FAILURE.value, 1, "big")
387
+            payload = int.to_bytes(_types.SSH_AGENT.FAILURE, 1, "big")
388 388
             self.send_to_client.extend(int.to_bytes(len(payload), 4, "big"))
389 389
             self.send_to_client.extend(payload)
390 390
         finally:
... ...
@@ -490,7 +490,7 @@ class StubbedSSHAgentSocket:
490 490
             iterable of 8-bit integers.
491 491
 
492 492
         """
493
-        yield _types.SSH_AGENT.EXTENSION_RESPONSE.value
493
+        yield _types.SSH_AGENT.EXTENSION_RESPONSE
494 494
         yield from ssh_agent.SSHAgentClient.string(b"query")
495 495
         extension_answers = [
496 496
             b"query",
... ...
@@ -518,9 +518,9 @@ class StubbedSSHAgentSocket:
518 518
 
519 519
         """
520 520
         if list_extended:
521
-            yield _types.SSH_AGENT.SUCCESS.value
521
+            yield _types.SSH_AGENT.SUCCESS
522 522
         else:
523
-            yield _types.SSH_AGENT.IDENTITIES_ANSWER.value
523
+            yield _types.SSH_AGENT.IDENTITIES_ANSWER
524 524
         signature_classes = [
525 525
             tests.data.SSHTestKeyDeterministicSignatureClass.SPEC,
526 526
         ]
... ...
@@ -576,13 +576,13 @@ class StubbedSSHAgentSocket:
576 576
             if key.public_key_data == key_blob:
577 577
                 if spec in key.expected_signatures:
578 578
                     return int.to_bytes(
579
-                        _types.SSH_AGENT.SIGN_RESPONSE.value, 1, "big"
579
+                        _types.SSH_AGENT.SIGN_RESPONSE, 1, "big"
580 580
                     ) + ssh_agent.SSHAgentClient.string(
581 581
                         key.expected_signatures[spec].signature
582 582
                     )
583 583
                 if try_rfc6979 and rfc6979 in key.expected_signatures:
584 584
                     return int.to_bytes(
585
-                        _types.SSH_AGENT.SIGN_RESPONSE.value, 1, "big"
585
+                        _types.SSH_AGENT.SIGN_RESPONSE, 1, "big"
586 586
                     ) + ssh_agent.SSHAgentClient.string(
587 587
                         key.expected_signatures[rfc6979].signature
588 588
                     )
... ...
@@ -444,10 +444,7 @@ class TestStubbedSSHAgentSocketRequests:
444 444
             orig_message: bytes | bytearray = bytearray(
445 445
                 agent.recv(message_length)
446 446
             )
447
-            assert (
448
-                _types.SSH_AGENT(orig_message[0])
449
-                == _types.SSH_AGENT.IDENTITIES_ANSWER
450
-            )
447
+            assert orig_message[0] == _types.SSH_AGENT.IDENTITIES_ANSWER
451 448
             identity_count = int.from_bytes(orig_message[1:5], "big")
452 449
             message = bytes(orig_message[5:])
453 450
             for _ in range(identity_count):
... ...
@@ -477,9 +474,7 @@ class TestStubbedSSHAgentSocketRequests:
477 474
             orig_message: bytes | bytearray = bytearray(
478 475
                 agent.recv(message_length)
479 476
             )
480
-            assert (
481
-                _types.SSH_AGENT(orig_message[0]) == _types.SSH_AGENT.SUCCESS
482
-            )
477
+            assert orig_message[0] == _types.SSH_AGENT.SUCCESS
483 478
             identity_count = int.from_bytes(orig_message[1:5], "big")
484 479
             message = bytes(orig_message[5:])
485 480
             for _ in range(identity_count):
... ...
@@ -160,9 +160,7 @@ class ListKeysAction(str, enum.Enum):
160 160
         if self == self.EMPTY:
161 161
             return []
162 162
         if self == self.FAIL:
163
-            raise ssh_agent.SSHAgentFailedError(
164
-                _types.SSH_AGENT.FAILURE.value, b""
165
-            )
163
+            raise ssh_agent.SSHAgentFailedError(_types.SSH_AGENT.FAILURE, b"")
166 164
         if self == self.FAIL_RUNTIME:
167 165
             raise ssh_agent.TrailingDataError()
168 166
         raise AssertionError()
... ...
@@ -188,9 +186,7 @@ class SignAction(str, enum.Enum):
188 186
         # matching.
189 187
         # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
190 188
         if self == self.FAIL:
191
-            raise ssh_agent.SSHAgentFailedError(
192
-                _types.SSH_AGENT.FAILURE.value, b""
193
-            )
189
+            raise ssh_agent.SSHAgentFailedError(_types.SSH_AGENT.FAILURE, b"")
194 190
         if self == self.FAIL_RUNTIME:
195 191
             raise ssh_agent.TrailingDataError()
196 192
         raise AssertionError()
... ...
@@ -344,7 +344,9 @@ class Parametrize(types.SimpleNamespace):
344 344
                 _types.SSH_AGENT.SUCCESS,
345 345
                 ssh_agent.SSHAgentFailedError,
346 346
                 re.escape(
347
-                    f"[Code {_types.SSH_AGENT.IDENTITIES_ANSWER.value}]"
347
+                    "[Code {:d}]".format(  # noqa: UP032
348
+                        int(_types.SSH_AGENT.IDENTITIES_ANSWER)
349
+                    )
348 350
                 ),
349 351
                 id="REQUEST_IDENTITIES-expect-SUCCESS",
350 352
             ),
... ...
@@ -1215,22 +1217,22 @@ class TestAgentErrorResponses:
1215 1217
             response_code = (
1216 1218
                 frozenset({_types.SSH_AGENT.SUCCESS})
1217 1219
                 if response_code is None
1218
-                else frozenset({_types.SSH_AGENT(response_code)})
1219
-                if isinstance(response_code, (int, _types.SSH_AGENT))
1220
-                else frozenset(map(_types.SSH_AGENT, response_code))
1220
+                else frozenset({response_code})
1221
+                if isinstance(response_code, int)
1222
+                else response_code
1221 1223
             )
1222 1224
             response_ = request_response_map.get(
1223 1225
                 self.Request(request_code, bytes(payload))
1224 1226
             ) or request_response_map.get(self.Request(request_code, None))
1225 1227
             if response_ is None:  # pragma: no cover [failsafe]
1226 1228
                 raise ssh_agent.SSHAgentFailedError(
1227
-                    _types.SSH_AGENT.FAILURE.value,
1229
+                    _types.SSH_AGENT.FAILURE,
1228 1230
                     "No prepared response for the request "
1229 1231
                     f"({request_code!r}, {payload!r})".encode(),
1230 1232
                 )
1231 1233
             code, response = response_
1232 1234
             if code not in response_code:
1233
-                raise ssh_agent.SSHAgentFailedError(code.value, response)
1235
+                raise ssh_agent.SSHAgentFailedError(code, response)
1234 1236
             return response
1235 1237
 
1236 1238
         return request
1237 1239