Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This file is dual licensed under the terms of the Apache License, Version
2# 2.0, and the BSD License. See the LICENSE file in the root of this repository
3# for complete details.
5from __future__ import annotations
7import binascii
8import enum
9import os
10import re
11import typing
12import warnings
13from dataclasses import dataclass
15from cryptography import utils
16from cryptography.exceptions import UnsupportedAlgorithm
17from cryptography.hazmat.primitives import hashes
18from cryptography.hazmat.primitives.asymmetric import (
19 dsa,
20 ec,
21 ed25519,
22 padding,
23 rsa,
24)
25from cryptography.hazmat.primitives.asymmetric import utils as asym_utils
26from cryptography.hazmat.primitives.ciphers import (
27 AEADDecryptionContext,
28 Cipher,
29 algorithms,
30 modes,
31)
32from cryptography.hazmat.primitives.serialization import (
33 Encoding,
34 KeySerializationEncryption,
35 NoEncryption,
36 PrivateFormat,
37 PublicFormat,
38 _KeySerializationEncryption,
39)
41try:
42 from bcrypt import kdf as _bcrypt_kdf
44 _bcrypt_supported = True
45except ImportError:
46 _bcrypt_supported = False
48 def _bcrypt_kdf(
49 password: bytes,
50 salt: bytes,
51 desired_key_bytes: int,
52 rounds: int,
53 ignore_few_rounds: bool = False,
54 ) -> bytes:
55 raise UnsupportedAlgorithm("Need bcrypt module")
58_SSH_ED25519 = b"ssh-ed25519"
59_SSH_RSA = b"ssh-rsa"
60_SSH_DSA = b"ssh-dss"
61_ECDSA_NISTP256 = b"ecdsa-sha2-nistp256"
62_ECDSA_NISTP384 = b"ecdsa-sha2-nistp384"
63_ECDSA_NISTP521 = b"ecdsa-sha2-nistp521"
64_CERT_SUFFIX = b"-cert-v01@openssh.com"
66# U2F application string suffixed pubkey
67_SK_SSH_ED25519 = b"sk-ssh-ed25519@openssh.com"
68_SK_SSH_ECDSA_NISTP256 = b"sk-ecdsa-sha2-nistp256@openssh.com"
70# These are not key types, only algorithms, so they cannot appear
71# as a public key type
72_SSH_RSA_SHA256 = b"rsa-sha2-256"
73_SSH_RSA_SHA512 = b"rsa-sha2-512"
75_SSH_PUBKEY_RC = re.compile(rb"\A(\S+)[ \t]+(\S+)")
76_SK_MAGIC = b"openssh-key-v1\0"
77_SK_START = b"-----BEGIN OPENSSH PRIVATE KEY-----"
78_SK_END = b"-----END OPENSSH PRIVATE KEY-----"
79_BCRYPT = b"bcrypt"
80_NONE = b"none"
81_DEFAULT_CIPHER = b"aes256-ctr"
82_DEFAULT_ROUNDS = 16
84# re is only way to work on bytes-like data
85_PEM_RC = re.compile(_SK_START + b"(.*?)" + _SK_END, re.DOTALL)
87# padding for max blocksize
88_PADDING = memoryview(bytearray(range(1, 1 + 16)))
91@dataclass
92class _SSHCipher:
93 alg: type[algorithms.AES]
94 key_len: int
95 mode: type[modes.CTR] | type[modes.CBC] | type[modes.GCM]
96 block_len: int
97 iv_len: int
98 tag_len: int | None
99 is_aead: bool
102# ciphers that are actually used in key wrapping
103_SSH_CIPHERS: dict[bytes, _SSHCipher] = {
104 b"aes256-ctr": _SSHCipher(
105 alg=algorithms.AES,
106 key_len=32,
107 mode=modes.CTR,
108 block_len=16,
109 iv_len=16,
110 tag_len=None,
111 is_aead=False,
112 ),
113 b"aes256-cbc": _SSHCipher(
114 alg=algorithms.AES,
115 key_len=32,
116 mode=modes.CBC,
117 block_len=16,
118 iv_len=16,
119 tag_len=None,
120 is_aead=False,
121 ),
122 b"aes256-gcm@openssh.com": _SSHCipher(
123 alg=algorithms.AES,
124 key_len=32,
125 mode=modes.GCM,
126 block_len=16,
127 iv_len=12,
128 tag_len=16,
129 is_aead=True,
130 ),
131}
133# map local curve name to key type
134_ECDSA_KEY_TYPE = {
135 "secp256r1": _ECDSA_NISTP256,
136 "secp384r1": _ECDSA_NISTP384,
137 "secp521r1": _ECDSA_NISTP521,
138}
141def _get_ssh_key_type(key: SSHPrivateKeyTypes | SSHPublicKeyTypes) -> bytes:
142 if isinstance(key, ec.EllipticCurvePrivateKey):
143 key_type = _ecdsa_key_type(key.public_key())
144 elif isinstance(key, ec.EllipticCurvePublicKey):
145 key_type = _ecdsa_key_type(key)
146 elif isinstance(key, (rsa.RSAPrivateKey, rsa.RSAPublicKey)):
147 key_type = _SSH_RSA
148 elif isinstance(key, (dsa.DSAPrivateKey, dsa.DSAPublicKey)):
149 key_type = _SSH_DSA
150 elif isinstance(
151 key, (ed25519.Ed25519PrivateKey, ed25519.Ed25519PublicKey)
152 ):
153 key_type = _SSH_ED25519
154 else:
155 raise ValueError("Unsupported key type")
157 return key_type
160def _ecdsa_key_type(public_key: ec.EllipticCurvePublicKey) -> bytes:
161 """Return SSH key_type and curve_name for private key."""
162 curve = public_key.curve
163 if curve.name not in _ECDSA_KEY_TYPE:
164 raise ValueError(
165 f"Unsupported curve for ssh private key: {curve.name!r}"
166 )
167 return _ECDSA_KEY_TYPE[curve.name]
170def _check_block_size(data: utils.Buffer, block_len: int) -> None:
171 """Require data to be full blocks"""
172 if not data or len(data) % block_len != 0:
173 raise ValueError("Corrupt data: missing padding")
176def _check_empty(data: utils.Buffer) -> None:
177 """All data should have been parsed."""
178 if data:
179 raise ValueError("Corrupt data: unparsed data")
182def _init_cipher(
183 ciphername: bytes,
184 password: bytes | None,
185 salt: bytes,
186 rounds: int,
187) -> Cipher[modes.CBC | modes.CTR | modes.GCM]:
188 """Generate key + iv and return cipher."""
189 if not password:
190 raise TypeError(
191 "Key is password-protected, but password was not provided."
192 )
194 ciph = _SSH_CIPHERS[ciphername]
195 seed = _bcrypt_kdf(
196 password, salt, ciph.key_len + ciph.iv_len, rounds, True
197 )
198 return Cipher(
199 ciph.alg(seed[: ciph.key_len]),
200 ciph.mode(seed[ciph.key_len :]),
201 )
204def _get_u32(data: memoryview) -> tuple[int, memoryview]:
205 """Uint32"""
206 if len(data) < 4:
207 raise ValueError("Invalid data")
208 return int.from_bytes(data[:4], byteorder="big"), data[4:]
211def _get_u64(data: memoryview) -> tuple[int, memoryview]:
212 """Uint64"""
213 if len(data) < 8:
214 raise ValueError("Invalid data")
215 return int.from_bytes(data[:8], byteorder="big"), data[8:]
218def _get_sshstr(data: memoryview) -> tuple[memoryview, memoryview]:
219 """Bytes with u32 length prefix"""
220 n, data = _get_u32(data)
221 if n > len(data):
222 raise ValueError("Invalid data")
223 return data[:n], data[n:]
226def _get_mpint(data: memoryview) -> tuple[int, memoryview]:
227 """Big integer."""
228 val, data = _get_sshstr(data)
229 if val and val[0] > 0x7F:
230 raise ValueError("Invalid data")
231 return int.from_bytes(val, "big"), data
234def _to_mpint(val: int) -> bytes:
235 """Storage format for signed bigint."""
236 if val < 0:
237 raise ValueError("negative mpint not allowed")
238 if not val:
239 return b""
240 nbytes = (val.bit_length() + 8) // 8
241 return utils.int_to_bytes(val, nbytes)
244class _FragList:
245 """Build recursive structure without data copy."""
247 flist: list[utils.Buffer]
249 def __init__(self, init: list[utils.Buffer] | None = None) -> None:
250 self.flist = []
251 if init:
252 self.flist.extend(init)
254 def put_raw(self, val: utils.Buffer) -> None:
255 """Add plain bytes"""
256 self.flist.append(val)
258 def put_u32(self, val: int) -> None:
259 """Big-endian uint32"""
260 self.flist.append(val.to_bytes(length=4, byteorder="big"))
262 def put_u64(self, val: int) -> None:
263 """Big-endian uint64"""
264 self.flist.append(val.to_bytes(length=8, byteorder="big"))
266 def put_sshstr(self, val: bytes | _FragList) -> None:
267 """Bytes prefixed with u32 length"""
268 if isinstance(val, (bytes, memoryview, bytearray)):
269 self.put_u32(len(val))
270 self.flist.append(val)
271 else:
272 self.put_u32(val.size())
273 self.flist.extend(val.flist)
275 def put_mpint(self, val: int) -> None:
276 """Big-endian bigint prefixed with u32 length"""
277 self.put_sshstr(_to_mpint(val))
279 def size(self) -> int:
280 """Current number of bytes"""
281 return sum(map(len, self.flist))
283 def render(self, dstbuf: memoryview, pos: int = 0) -> int:
284 """Write into bytearray"""
285 for frag in self.flist:
286 flen = len(frag)
287 start, pos = pos, pos + flen
288 dstbuf[start:pos] = frag
289 return pos
291 def tobytes(self) -> bytes:
292 """Return as bytes"""
293 buf = memoryview(bytearray(self.size()))
294 self.render(buf)
295 return buf.tobytes()
298class _SSHFormatRSA:
299 """Format for RSA keys.
301 Public:
302 mpint e, n
303 Private:
304 mpint n, e, d, iqmp, p, q
305 """
307 def get_public(
308 self, data: memoryview
309 ) -> tuple[tuple[int, int], memoryview]:
310 """RSA public fields"""
311 e, data = _get_mpint(data)
312 n, data = _get_mpint(data)
313 return (e, n), data
315 def load_public(
316 self, data: memoryview
317 ) -> tuple[rsa.RSAPublicKey, memoryview]:
318 """Make RSA public key from data."""
319 (e, n), data = self.get_public(data)
320 public_numbers = rsa.RSAPublicNumbers(e, n)
321 public_key = public_numbers.public_key()
322 return public_key, data
324 def load_private(
325 self, data: memoryview, pubfields, unsafe_skip_rsa_key_validation: bool
326 ) -> tuple[rsa.RSAPrivateKey, memoryview]:
327 """Make RSA private key from data."""
328 n, data = _get_mpint(data)
329 e, data = _get_mpint(data)
330 d, data = _get_mpint(data)
331 iqmp, data = _get_mpint(data)
332 p, data = _get_mpint(data)
333 q, data = _get_mpint(data)
335 if (e, n) != pubfields:
336 raise ValueError("Corrupt data: rsa field mismatch")
337 dmp1 = rsa.rsa_crt_dmp1(d, p)
338 dmq1 = rsa.rsa_crt_dmq1(d, q)
339 public_numbers = rsa.RSAPublicNumbers(e, n)
340 private_numbers = rsa.RSAPrivateNumbers(
341 p, q, d, dmp1, dmq1, iqmp, public_numbers
342 )
343 private_key = private_numbers.private_key(
344 unsafe_skip_rsa_key_validation=unsafe_skip_rsa_key_validation
345 )
346 return private_key, data
348 def encode_public(
349 self, public_key: rsa.RSAPublicKey, f_pub: _FragList
350 ) -> None:
351 """Write RSA public key"""
352 pubn = public_key.public_numbers()
353 f_pub.put_mpint(pubn.e)
354 f_pub.put_mpint(pubn.n)
356 def encode_private(
357 self, private_key: rsa.RSAPrivateKey, f_priv: _FragList
358 ) -> None:
359 """Write RSA private key"""
360 private_numbers = private_key.private_numbers()
361 public_numbers = private_numbers.public_numbers
363 f_priv.put_mpint(public_numbers.n)
364 f_priv.put_mpint(public_numbers.e)
366 f_priv.put_mpint(private_numbers.d)
367 f_priv.put_mpint(private_numbers.iqmp)
368 f_priv.put_mpint(private_numbers.p)
369 f_priv.put_mpint(private_numbers.q)
372class _SSHFormatDSA:
373 """Format for DSA keys.
375 Public:
376 mpint p, q, g, y
377 Private:
378 mpint p, q, g, y, x
379 """
381 def get_public(self, data: memoryview) -> tuple[tuple, memoryview]:
382 """DSA public fields"""
383 p, data = _get_mpint(data)
384 q, data = _get_mpint(data)
385 g, data = _get_mpint(data)
386 y, data = _get_mpint(data)
387 return (p, q, g, y), data
389 def load_public(
390 self, data: memoryview
391 ) -> tuple[dsa.DSAPublicKey, memoryview]:
392 """Make DSA public key from data."""
393 (p, q, g, y), data = self.get_public(data)
394 parameter_numbers = dsa.DSAParameterNumbers(p, q, g)
395 public_numbers = dsa.DSAPublicNumbers(y, parameter_numbers)
396 self._validate(public_numbers)
397 public_key = public_numbers.public_key()
398 return public_key, data
400 def load_private(
401 self, data: memoryview, pubfields, unsafe_skip_rsa_key_validation: bool
402 ) -> tuple[dsa.DSAPrivateKey, memoryview]:
403 """Make DSA private key from data."""
404 (p, q, g, y), data = self.get_public(data)
405 x, data = _get_mpint(data)
407 if (p, q, g, y) != pubfields:
408 raise ValueError("Corrupt data: dsa field mismatch")
409 parameter_numbers = dsa.DSAParameterNumbers(p, q, g)
410 public_numbers = dsa.DSAPublicNumbers(y, parameter_numbers)
411 self._validate(public_numbers)
412 private_numbers = dsa.DSAPrivateNumbers(x, public_numbers)
413 private_key = private_numbers.private_key()
414 return private_key, data
416 def encode_public(
417 self, public_key: dsa.DSAPublicKey, f_pub: _FragList
418 ) -> None:
419 """Write DSA public key"""
420 public_numbers = public_key.public_numbers()
421 parameter_numbers = public_numbers.parameter_numbers
422 self._validate(public_numbers)
424 f_pub.put_mpint(parameter_numbers.p)
425 f_pub.put_mpint(parameter_numbers.q)
426 f_pub.put_mpint(parameter_numbers.g)
427 f_pub.put_mpint(public_numbers.y)
429 def encode_private(
430 self, private_key: dsa.DSAPrivateKey, f_priv: _FragList
431 ) -> None:
432 """Write DSA private key"""
433 self.encode_public(private_key.public_key(), f_priv)
434 f_priv.put_mpint(private_key.private_numbers().x)
436 def _validate(self, public_numbers: dsa.DSAPublicNumbers) -> None:
437 parameter_numbers = public_numbers.parameter_numbers
438 if parameter_numbers.p.bit_length() != 1024:
439 raise ValueError("SSH supports only 1024 bit DSA keys")
442class _SSHFormatECDSA:
443 """Format for ECDSA keys.
445 Public:
446 str curve
447 bytes point
448 Private:
449 str curve
450 bytes point
451 mpint secret
452 """
454 def __init__(self, ssh_curve_name: bytes, curve: ec.EllipticCurve):
455 self.ssh_curve_name = ssh_curve_name
456 self.curve = curve
458 def get_public(
459 self, data: memoryview
460 ) -> tuple[tuple[memoryview, memoryview], memoryview]:
461 """ECDSA public fields"""
462 curve, data = _get_sshstr(data)
463 point, data = _get_sshstr(data)
464 if curve != self.ssh_curve_name:
465 raise ValueError("Curve name mismatch")
466 if len(point) == 0:
467 raise ValueError("Invalid EC point: empty data")
468 if point[0] != 4:
469 raise NotImplementedError("Need uncompressed point")
470 return (curve, point), data
472 def load_public(
473 self, data: memoryview
474 ) -> tuple[ec.EllipticCurvePublicKey, memoryview]:
475 """Make ECDSA public key from data."""
476 (_, point), data = self.get_public(data)
477 public_key = ec.EllipticCurvePublicKey.from_encoded_point(
478 self.curve, point.tobytes()
479 )
480 return public_key, data
482 def load_private(
483 self, data: memoryview, pubfields, unsafe_skip_rsa_key_validation: bool
484 ) -> tuple[ec.EllipticCurvePrivateKey, memoryview]:
485 """Make ECDSA private key from data."""
486 (curve_name, point), data = self.get_public(data)
487 secret, data = _get_mpint(data)
489 if (curve_name, point) != pubfields:
490 raise ValueError("Corrupt data: ecdsa field mismatch")
491 private_key = ec.derive_private_key(secret, self.curve)
492 return private_key, data
494 def encode_public(
495 self, public_key: ec.EllipticCurvePublicKey, f_pub: _FragList
496 ) -> None:
497 """Write ECDSA public key"""
498 point = public_key.public_bytes(
499 Encoding.X962, PublicFormat.UncompressedPoint
500 )
501 f_pub.put_sshstr(self.ssh_curve_name)
502 f_pub.put_sshstr(point)
504 def encode_private(
505 self, private_key: ec.EllipticCurvePrivateKey, f_priv: _FragList
506 ) -> None:
507 """Write ECDSA private key"""
508 public_key = private_key.public_key()
509 private_numbers = private_key.private_numbers()
511 self.encode_public(public_key, f_priv)
512 f_priv.put_mpint(private_numbers.private_value)
515class _SSHFormatEd25519:
516 """Format for Ed25519 keys.
518 Public:
519 bytes point
520 Private:
521 bytes point
522 bytes secret_and_point
523 """
525 def get_public(
526 self, data: memoryview
527 ) -> tuple[tuple[memoryview], memoryview]:
528 """Ed25519 public fields"""
529 point, data = _get_sshstr(data)
530 return (point,), data
532 def load_public(
533 self, data: memoryview
534 ) -> tuple[ed25519.Ed25519PublicKey, memoryview]:
535 """Make Ed25519 public key from data."""
536 (point,), data = self.get_public(data)
537 public_key = ed25519.Ed25519PublicKey.from_public_bytes(
538 point.tobytes()
539 )
540 return public_key, data
542 def load_private(
543 self, data: memoryview, pubfields, unsafe_skip_rsa_key_validation: bool
544 ) -> tuple[ed25519.Ed25519PrivateKey, memoryview]:
545 """Make Ed25519 private key from data."""
546 (point,), data = self.get_public(data)
547 keypair, data = _get_sshstr(data)
549 secret = keypair[:32]
550 point2 = keypair[32:]
551 if point != point2 or (point,) != pubfields:
552 raise ValueError("Corrupt data: ed25519 field mismatch")
553 private_key = ed25519.Ed25519PrivateKey.from_private_bytes(secret)
554 return private_key, data
556 def encode_public(
557 self, public_key: ed25519.Ed25519PublicKey, f_pub: _FragList
558 ) -> None:
559 """Write Ed25519 public key"""
560 raw_public_key = public_key.public_bytes(
561 Encoding.Raw, PublicFormat.Raw
562 )
563 f_pub.put_sshstr(raw_public_key)
565 def encode_private(
566 self, private_key: ed25519.Ed25519PrivateKey, f_priv: _FragList
567 ) -> None:
568 """Write Ed25519 private key"""
569 public_key = private_key.public_key()
570 raw_private_key = private_key.private_bytes(
571 Encoding.Raw, PrivateFormat.Raw, NoEncryption()
572 )
573 raw_public_key = public_key.public_bytes(
574 Encoding.Raw, PublicFormat.Raw
575 )
576 f_keypair = _FragList([raw_private_key, raw_public_key])
578 self.encode_public(public_key, f_priv)
579 f_priv.put_sshstr(f_keypair)
582def load_application(data) -> tuple[memoryview, memoryview]:
583 """
584 U2F application strings
585 """
586 application, data = _get_sshstr(data)
587 if not application.tobytes().startswith(b"ssh:"):
588 raise ValueError(
589 "U2F application string does not start with b'ssh:' "
590 f"({application})"
591 )
592 return application, data
595class _SSHFormatSKEd25519:
596 """
597 The format of a sk-ssh-ed25519@openssh.com public key is:
599 string "sk-ssh-ed25519@openssh.com"
600 string public key
601 string application (user-specified, but typically "ssh:")
602 """
604 def load_public(
605 self, data: memoryview
606 ) -> tuple[ed25519.Ed25519PublicKey, memoryview]:
607 """Make Ed25519 public key from data."""
608 public_key, data = _lookup_kformat(_SSH_ED25519).load_public(data)
609 _, data = load_application(data)
610 return public_key, data
612 def get_public(self, data: memoryview) -> typing.NoReturn:
613 # Confusingly `get_public` is an entry point used by private key
614 # loading.
615 raise UnsupportedAlgorithm(
616 "sk-ssh-ed25519 private keys cannot be loaded"
617 )
620class _SSHFormatSKECDSA:
621 """
622 The format of a sk-ecdsa-sha2-nistp256@openssh.com public key is:
624 string "sk-ecdsa-sha2-nistp256@openssh.com"
625 string curve name
626 ec_point Q
627 string application (user-specified, but typically "ssh:")
628 """
630 def load_public(
631 self, data: memoryview
632 ) -> tuple[ec.EllipticCurvePublicKey, memoryview]:
633 """Make ECDSA public key from data."""
634 public_key, data = _lookup_kformat(_ECDSA_NISTP256).load_public(data)
635 _, data = load_application(data)
636 return public_key, data
638 def get_public(self, data: memoryview) -> typing.NoReturn:
639 # Confusingly `get_public` is an entry point used by private key
640 # loading.
641 raise UnsupportedAlgorithm(
642 "sk-ecdsa-sha2-nistp256 private keys cannot be loaded"
643 )
646_KEY_FORMATS = {
647 _SSH_RSA: _SSHFormatRSA(),
648 _SSH_DSA: _SSHFormatDSA(),
649 _SSH_ED25519: _SSHFormatEd25519(),
650 _ECDSA_NISTP256: _SSHFormatECDSA(b"nistp256", ec.SECP256R1()),
651 _ECDSA_NISTP384: _SSHFormatECDSA(b"nistp384", ec.SECP384R1()),
652 _ECDSA_NISTP521: _SSHFormatECDSA(b"nistp521", ec.SECP521R1()),
653 _SK_SSH_ED25519: _SSHFormatSKEd25519(),
654 _SK_SSH_ECDSA_NISTP256: _SSHFormatSKECDSA(),
655}
658def _lookup_kformat(key_type: utils.Buffer):
659 """Return valid format or throw error"""
660 if not isinstance(key_type, bytes):
661 key_type = memoryview(key_type).tobytes()
662 if key_type in _KEY_FORMATS:
663 return _KEY_FORMATS[key_type]
664 raise UnsupportedAlgorithm(f"Unsupported key type: {key_type!r}")
667SSHPrivateKeyTypes = typing.Union[
668 ec.EllipticCurvePrivateKey,
669 rsa.RSAPrivateKey,
670 dsa.DSAPrivateKey,
671 ed25519.Ed25519PrivateKey,
672]
675def load_ssh_private_key(
676 data: utils.Buffer,
677 password: bytes | None,
678 backend: typing.Any = None,
679 *,
680 unsafe_skip_rsa_key_validation: bool = False,
681) -> SSHPrivateKeyTypes:
682 """Load private key from OpenSSH custom encoding."""
683 utils._check_byteslike("data", data)
684 if password is not None:
685 utils._check_bytes("password", password)
687 m = _PEM_RC.search(data)
688 if not m:
689 raise ValueError("Not OpenSSH private key format")
690 p1 = m.start(1)
691 p2 = m.end(1)
692 data = binascii.a2b_base64(memoryview(data)[p1:p2])
693 if not data.startswith(_SK_MAGIC):
694 raise ValueError("Not OpenSSH private key format")
695 data = memoryview(data)[len(_SK_MAGIC) :]
697 # parse header
698 ciphername, data = _get_sshstr(data)
699 kdfname, data = _get_sshstr(data)
700 kdfoptions, data = _get_sshstr(data)
701 nkeys, data = _get_u32(data)
702 if nkeys != 1:
703 raise ValueError("Only one key supported")
705 # load public key data
706 pubdata, data = _get_sshstr(data)
707 pub_key_type, pubdata = _get_sshstr(pubdata)
708 kformat = _lookup_kformat(pub_key_type)
709 pubfields, pubdata = kformat.get_public(pubdata)
710 _check_empty(pubdata)
712 if ciphername != _NONE or kdfname != _NONE:
713 ciphername_bytes = ciphername.tobytes()
714 if ciphername_bytes not in _SSH_CIPHERS:
715 raise UnsupportedAlgorithm(
716 f"Unsupported cipher: {ciphername_bytes!r}"
717 )
718 if kdfname != _BCRYPT:
719 raise UnsupportedAlgorithm(f"Unsupported KDF: {kdfname!r}")
720 blklen = _SSH_CIPHERS[ciphername_bytes].block_len
721 tag_len = _SSH_CIPHERS[ciphername_bytes].tag_len
722 # load secret data
723 edata, data = _get_sshstr(data)
724 # see https://bugzilla.mindrot.org/show_bug.cgi?id=3553 for
725 # information about how OpenSSH handles AEAD tags
726 if _SSH_CIPHERS[ciphername_bytes].is_aead:
727 tag = bytes(data)
728 if len(tag) != tag_len:
729 raise ValueError("Corrupt data: invalid tag length for cipher")
730 else:
731 _check_empty(data)
732 _check_block_size(edata, blklen)
733 salt, kbuf = _get_sshstr(kdfoptions)
734 rounds, kbuf = _get_u32(kbuf)
735 _check_empty(kbuf)
736 ciph = _init_cipher(ciphername_bytes, password, salt.tobytes(), rounds)
737 dec = ciph.decryptor()
738 edata = memoryview(dec.update(edata))
739 if _SSH_CIPHERS[ciphername_bytes].is_aead:
740 assert isinstance(dec, AEADDecryptionContext)
741 _check_empty(dec.finalize_with_tag(tag))
742 else:
743 # _check_block_size requires data to be a full block so there
744 # should be no output from finalize
745 _check_empty(dec.finalize())
746 else:
747 if password:
748 raise TypeError(
749 "Password was given but private key is not encrypted."
750 )
751 # load secret data
752 edata, data = _get_sshstr(data)
753 _check_empty(data)
754 blklen = 8
755 _check_block_size(edata, blklen)
756 ck1, edata = _get_u32(edata)
757 ck2, edata = _get_u32(edata)
758 if ck1 != ck2:
759 raise ValueError("Corrupt data: broken checksum")
761 # load per-key struct
762 key_type, edata = _get_sshstr(edata)
763 if key_type != pub_key_type:
764 raise ValueError("Corrupt data: key type mismatch")
765 private_key, edata = kformat.load_private(
766 edata,
767 pubfields,
768 unsafe_skip_rsa_key_validation=unsafe_skip_rsa_key_validation,
769 )
770 # We don't use the comment
771 _, edata = _get_sshstr(edata)
773 # yes, SSH does padding check *after* all other parsing is done.
774 # need to follow as it writes zero-byte padding too.
775 if edata != _PADDING[: len(edata)]:
776 raise ValueError("Corrupt data: invalid padding")
778 if isinstance(private_key, dsa.DSAPrivateKey):
779 warnings.warn(
780 "SSH DSA keys are deprecated and will be removed in a future "
781 "release.",
782 utils.DeprecatedIn40,
783 stacklevel=2,
784 )
786 return private_key
789def _serialize_ssh_private_key(
790 private_key: SSHPrivateKeyTypes,
791 password: bytes,
792 encryption_algorithm: KeySerializationEncryption,
793) -> bytes:
794 """Serialize private key with OpenSSH custom encoding."""
795 utils._check_bytes("password", password)
796 if isinstance(private_key, dsa.DSAPrivateKey):
797 warnings.warn(
798 "SSH DSA key support is deprecated and will be "
799 "removed in a future release",
800 utils.DeprecatedIn40,
801 stacklevel=4,
802 )
804 key_type = _get_ssh_key_type(private_key)
805 kformat = _lookup_kformat(key_type)
807 # setup parameters
808 f_kdfoptions = _FragList()
809 if password:
810 ciphername = _DEFAULT_CIPHER
811 blklen = _SSH_CIPHERS[ciphername].block_len
812 kdfname = _BCRYPT
813 rounds = _DEFAULT_ROUNDS
814 if (
815 isinstance(encryption_algorithm, _KeySerializationEncryption)
816 and encryption_algorithm._kdf_rounds is not None
817 ):
818 rounds = encryption_algorithm._kdf_rounds
819 salt = os.urandom(16)
820 f_kdfoptions.put_sshstr(salt)
821 f_kdfoptions.put_u32(rounds)
822 ciph = _init_cipher(ciphername, password, salt, rounds)
823 else:
824 ciphername = kdfname = _NONE
825 blklen = 8
826 ciph = None
827 nkeys = 1
828 checkval = os.urandom(4)
829 comment = b""
831 # encode public and private parts together
832 f_public_key = _FragList()
833 f_public_key.put_sshstr(key_type)
834 kformat.encode_public(private_key.public_key(), f_public_key)
836 f_secrets = _FragList([checkval, checkval])
837 f_secrets.put_sshstr(key_type)
838 kformat.encode_private(private_key, f_secrets)
839 f_secrets.put_sshstr(comment)
840 f_secrets.put_raw(_PADDING[: blklen - (f_secrets.size() % blklen)])
842 # top-level structure
843 f_main = _FragList()
844 f_main.put_raw(_SK_MAGIC)
845 f_main.put_sshstr(ciphername)
846 f_main.put_sshstr(kdfname)
847 f_main.put_sshstr(f_kdfoptions)
848 f_main.put_u32(nkeys)
849 f_main.put_sshstr(f_public_key)
850 f_main.put_sshstr(f_secrets)
852 # copy result info bytearray
853 slen = f_secrets.size()
854 mlen = f_main.size()
855 buf = memoryview(bytearray(mlen + blklen))
856 f_main.render(buf)
857 ofs = mlen - slen
859 # encrypt in-place
860 if ciph is not None:
861 ciph.encryptor().update_into(buf[ofs:mlen], buf[ofs:])
863 return bytes(buf[:mlen])
866SSHPublicKeyTypes = typing.Union[
867 ec.EllipticCurvePublicKey,
868 rsa.RSAPublicKey,
869 dsa.DSAPublicKey,
870 ed25519.Ed25519PublicKey,
871]
873SSHCertPublicKeyTypes = typing.Union[
874 ec.EllipticCurvePublicKey,
875 rsa.RSAPublicKey,
876 ed25519.Ed25519PublicKey,
877]
880class SSHCertificateType(enum.Enum):
881 USER = 1
882 HOST = 2
885class SSHCertificate:
886 def __init__(
887 self,
888 _nonce: memoryview,
889 _public_key: SSHPublicKeyTypes,
890 _serial: int,
891 _cctype: int,
892 _key_id: memoryview,
893 _valid_principals: list[bytes],
894 _valid_after: int,
895 _valid_before: int,
896 _critical_options: dict[bytes, bytes],
897 _extensions: dict[bytes, bytes],
898 _sig_type: memoryview,
899 _sig_key: memoryview,
900 _inner_sig_type: memoryview,
901 _signature: memoryview,
902 _tbs_cert_body: memoryview,
903 _cert_key_type: bytes,
904 _cert_body: memoryview,
905 ):
906 self._nonce = _nonce
907 self._public_key = _public_key
908 self._serial = _serial
909 try:
910 self._type = SSHCertificateType(_cctype)
911 except ValueError:
912 raise ValueError("Invalid certificate type")
913 self._key_id = _key_id
914 self._valid_principals = _valid_principals
915 self._valid_after = _valid_after
916 self._valid_before = _valid_before
917 self._critical_options = _critical_options
918 self._extensions = _extensions
919 self._sig_type = _sig_type
920 self._sig_key = _sig_key
921 self._inner_sig_type = _inner_sig_type
922 self._signature = _signature
923 self._cert_key_type = _cert_key_type
924 self._cert_body = _cert_body
925 self._tbs_cert_body = _tbs_cert_body
927 @property
928 def nonce(self) -> bytes:
929 return bytes(self._nonce)
931 def public_key(self) -> SSHCertPublicKeyTypes:
932 # make mypy happy until we remove DSA support entirely and
933 # the underlying union won't have a disallowed type
934 return typing.cast(SSHCertPublicKeyTypes, self._public_key)
936 @property
937 def serial(self) -> int:
938 return self._serial
940 @property
941 def type(self) -> SSHCertificateType:
942 return self._type
944 @property
945 def key_id(self) -> bytes:
946 return bytes(self._key_id)
948 @property
949 def valid_principals(self) -> list[bytes]:
950 return self._valid_principals
952 @property
953 def valid_before(self) -> int:
954 return self._valid_before
956 @property
957 def valid_after(self) -> int:
958 return self._valid_after
960 @property
961 def critical_options(self) -> dict[bytes, bytes]:
962 return self._critical_options
964 @property
965 def extensions(self) -> dict[bytes, bytes]:
966 return self._extensions
968 def signature_key(self) -> SSHCertPublicKeyTypes:
969 sigformat = _lookup_kformat(self._sig_type)
970 signature_key, sigkey_rest = sigformat.load_public(self._sig_key)
971 _check_empty(sigkey_rest)
972 return signature_key
974 def public_bytes(self) -> bytes:
975 return (
976 bytes(self._cert_key_type)
977 + b" "
978 + binascii.b2a_base64(bytes(self._cert_body), newline=False)
979 )
981 def verify_cert_signature(self) -> None:
982 signature_key = self.signature_key()
983 if isinstance(signature_key, ed25519.Ed25519PublicKey):
984 signature_key.verify(
985 bytes(self._signature), bytes(self._tbs_cert_body)
986 )
987 elif isinstance(signature_key, ec.EllipticCurvePublicKey):
988 # The signature is encoded as a pair of big-endian integers
989 r, data = _get_mpint(self._signature)
990 s, data = _get_mpint(data)
991 _check_empty(data)
992 computed_sig = asym_utils.encode_dss_signature(r, s)
993 hash_alg = _get_ec_hash_alg(signature_key.curve)
994 signature_key.verify(
995 computed_sig, bytes(self._tbs_cert_body), ec.ECDSA(hash_alg)
996 )
997 else:
998 assert isinstance(signature_key, rsa.RSAPublicKey)
999 if self._inner_sig_type == _SSH_RSA:
1000 hash_alg = hashes.SHA1()
1001 elif self._inner_sig_type == _SSH_RSA_SHA256:
1002 hash_alg = hashes.SHA256()
1003 else:
1004 assert self._inner_sig_type == _SSH_RSA_SHA512
1005 hash_alg = hashes.SHA512()
1006 signature_key.verify(
1007 bytes(self._signature),
1008 bytes(self._tbs_cert_body),
1009 padding.PKCS1v15(),
1010 hash_alg,
1011 )
1014def _get_ec_hash_alg(curve: ec.EllipticCurve) -> hashes.HashAlgorithm:
1015 if isinstance(curve, ec.SECP256R1):
1016 return hashes.SHA256()
1017 elif isinstance(curve, ec.SECP384R1):
1018 return hashes.SHA384()
1019 else:
1020 assert isinstance(curve, ec.SECP521R1)
1021 return hashes.SHA512()
1024def _load_ssh_public_identity(
1025 data: utils.Buffer,
1026 _legacy_dsa_allowed=False,
1027) -> SSHCertificate | SSHPublicKeyTypes:
1028 utils._check_byteslike("data", data)
1030 m = _SSH_PUBKEY_RC.match(data)
1031 if not m:
1032 raise ValueError("Invalid line format")
1033 key_type = orig_key_type = m.group(1)
1034 key_body = m.group(2)
1035 with_cert = False
1036 if key_type.endswith(_CERT_SUFFIX):
1037 with_cert = True
1038 key_type = key_type[: -len(_CERT_SUFFIX)]
1039 if key_type == _SSH_DSA and not _legacy_dsa_allowed:
1040 raise UnsupportedAlgorithm(
1041 "DSA keys aren't supported in SSH certificates"
1042 )
1043 kformat = _lookup_kformat(key_type)
1045 try:
1046 rest = memoryview(binascii.a2b_base64(key_body))
1047 except (TypeError, binascii.Error):
1048 raise ValueError("Invalid format")
1050 if with_cert:
1051 cert_body = rest
1052 inner_key_type, rest = _get_sshstr(rest)
1053 if inner_key_type != orig_key_type:
1054 raise ValueError("Invalid key format")
1055 if with_cert:
1056 nonce, rest = _get_sshstr(rest)
1057 public_key, rest = kformat.load_public(rest)
1058 if with_cert:
1059 serial, rest = _get_u64(rest)
1060 cctype, rest = _get_u32(rest)
1061 key_id, rest = _get_sshstr(rest)
1062 principals, rest = _get_sshstr(rest)
1063 valid_principals = []
1064 while principals:
1065 principal, principals = _get_sshstr(principals)
1066 valid_principals.append(bytes(principal))
1067 valid_after, rest = _get_u64(rest)
1068 valid_before, rest = _get_u64(rest)
1069 crit_options, rest = _get_sshstr(rest)
1070 critical_options = _parse_exts_opts(crit_options)
1071 exts, rest = _get_sshstr(rest)
1072 extensions = _parse_exts_opts(exts)
1073 # Get the reserved field, which is unused.
1074 _, rest = _get_sshstr(rest)
1075 sig_key_raw, rest = _get_sshstr(rest)
1076 sig_type, sig_key = _get_sshstr(sig_key_raw)
1077 if sig_type == _SSH_DSA and not _legacy_dsa_allowed:
1078 raise UnsupportedAlgorithm(
1079 "DSA signatures aren't supported in SSH certificates"
1080 )
1081 # Get the entire cert body and subtract the signature
1082 tbs_cert_body = cert_body[: -len(rest)]
1083 signature_raw, rest = _get_sshstr(rest)
1084 _check_empty(rest)
1085 inner_sig_type, sig_rest = _get_sshstr(signature_raw)
1086 # RSA certs can have multiple algorithm types
1087 if (
1088 sig_type == _SSH_RSA
1089 and inner_sig_type
1090 not in [_SSH_RSA_SHA256, _SSH_RSA_SHA512, _SSH_RSA]
1091 ) or (sig_type != _SSH_RSA and inner_sig_type != sig_type):
1092 raise ValueError("Signature key type does not match")
1093 signature, sig_rest = _get_sshstr(sig_rest)
1094 _check_empty(sig_rest)
1095 return SSHCertificate(
1096 nonce,
1097 public_key,
1098 serial,
1099 cctype,
1100 key_id,
1101 valid_principals,
1102 valid_after,
1103 valid_before,
1104 critical_options,
1105 extensions,
1106 sig_type,
1107 sig_key,
1108 inner_sig_type,
1109 signature,
1110 tbs_cert_body,
1111 orig_key_type,
1112 cert_body,
1113 )
1114 else:
1115 _check_empty(rest)
1116 return public_key
1119def load_ssh_public_identity(
1120 data: utils.Buffer,
1121) -> SSHCertificate | SSHPublicKeyTypes:
1122 return _load_ssh_public_identity(data)
1125def _parse_exts_opts(exts_opts: memoryview) -> dict[bytes, bytes]:
1126 result: dict[bytes, bytes] = {}
1127 last_name = None
1128 while exts_opts:
1129 name, exts_opts = _get_sshstr(exts_opts)
1130 bname: bytes = bytes(name)
1131 if bname in result:
1132 raise ValueError("Duplicate name")
1133 if last_name is not None and bname < last_name:
1134 raise ValueError("Fields not lexically sorted")
1135 value, exts_opts = _get_sshstr(exts_opts)
1136 if len(value) > 0:
1137 value, extra = _get_sshstr(value)
1138 if len(extra) > 0:
1139 raise ValueError("Unexpected extra data after value")
1140 result[bname] = bytes(value)
1141 last_name = bname
1142 return result
1145def ssh_key_fingerprint(
1146 key: SSHPublicKeyTypes,
1147 hash_algorithm: hashes.MD5 | hashes.SHA1 | hashes.SHA256,
1148) -> bytes:
1149 if not isinstance(
1150 hash_algorithm,
1151 (hashes.MD5, hashes.SHA1, hashes.SHA256),
1152 ):
1153 raise TypeError("hash_algorithm must be either MD5, SHA1, or SHA256")
1155 key_type = _get_ssh_key_type(key)
1156 kformat = _lookup_kformat(key_type)
1158 f_pub = _FragList()
1159 f_pub.put_sshstr(key_type)
1160 kformat.encode_public(key, f_pub)
1162 ssh_binary_data = f_pub.tobytes()
1164 # Hash the binary data
1165 hash_obj = hashes.Hash(hash_algorithm)
1166 hash_obj.update(ssh_binary_data)
1167 return hash_obj.finalize()
1170def load_ssh_public_key(
1171 data: utils.Buffer, backend: typing.Any = None
1172) -> SSHPublicKeyTypes:
1173 cert_or_key = _load_ssh_public_identity(data, _legacy_dsa_allowed=True)
1174 public_key: SSHPublicKeyTypes
1175 if isinstance(cert_or_key, SSHCertificate):
1176 public_key = cert_or_key.public_key()
1177 else:
1178 public_key = cert_or_key
1180 if isinstance(public_key, dsa.DSAPublicKey):
1181 warnings.warn(
1182 "SSH DSA keys are deprecated and will be removed in a future "
1183 "release.",
1184 utils.DeprecatedIn40,
1185 stacklevel=2,
1186 )
1187 return public_key
1190def serialize_ssh_public_key(public_key: SSHPublicKeyTypes) -> bytes:
1191 """One-line public key format for OpenSSH"""
1192 if isinstance(public_key, dsa.DSAPublicKey):
1193 warnings.warn(
1194 "SSH DSA key support is deprecated and will be "
1195 "removed in a future release",
1196 utils.DeprecatedIn40,
1197 stacklevel=4,
1198 )
1199 key_type = _get_ssh_key_type(public_key)
1200 kformat = _lookup_kformat(key_type)
1202 f_pub = _FragList()
1203 f_pub.put_sshstr(key_type)
1204 kformat.encode_public(public_key, f_pub)
1206 pub = binascii.b2a_base64(f_pub.tobytes()).strip()
1207 return b"".join([key_type, b" ", pub])
1210SSHCertPrivateKeyTypes = typing.Union[
1211 ec.EllipticCurvePrivateKey,
1212 rsa.RSAPrivateKey,
1213 ed25519.Ed25519PrivateKey,
1214]
1217# This is an undocumented limit enforced in the openssh codebase for sshd and
1218# ssh-keygen, but it is undefined in the ssh certificates spec.
1219_SSHKEY_CERT_MAX_PRINCIPALS = 256
1222class SSHCertificateBuilder:
1223 def __init__(
1224 self,
1225 _public_key: SSHCertPublicKeyTypes | None = None,
1226 _serial: int | None = None,
1227 _type: SSHCertificateType | None = None,
1228 _key_id: bytes | None = None,
1229 _valid_principals: list[bytes] = [],
1230 _valid_for_all_principals: bool = False,
1231 _valid_before: int | None = None,
1232 _valid_after: int | None = None,
1233 _critical_options: list[tuple[bytes, bytes]] = [],
1234 _extensions: list[tuple[bytes, bytes]] = [],
1235 ):
1236 self._public_key = _public_key
1237 self._serial = _serial
1238 self._type = _type
1239 self._key_id = _key_id
1240 self._valid_principals = _valid_principals
1241 self._valid_for_all_principals = _valid_for_all_principals
1242 self._valid_before = _valid_before
1243 self._valid_after = _valid_after
1244 self._critical_options = _critical_options
1245 self._extensions = _extensions
1247 def public_key(
1248 self, public_key: SSHCertPublicKeyTypes
1249 ) -> SSHCertificateBuilder:
1250 if not isinstance(
1251 public_key,
1252 (
1253 ec.EllipticCurvePublicKey,
1254 rsa.RSAPublicKey,
1255 ed25519.Ed25519PublicKey,
1256 ),
1257 ):
1258 raise TypeError("Unsupported key type")
1259 if self._public_key is not None:
1260 raise ValueError("public_key already set")
1262 return SSHCertificateBuilder(
1263 _public_key=public_key,
1264 _serial=self._serial,
1265 _type=self._type,
1266 _key_id=self._key_id,
1267 _valid_principals=self._valid_principals,
1268 _valid_for_all_principals=self._valid_for_all_principals,
1269 _valid_before=self._valid_before,
1270 _valid_after=self._valid_after,
1271 _critical_options=self._critical_options,
1272 _extensions=self._extensions,
1273 )
1275 def serial(self, serial: int) -> SSHCertificateBuilder:
1276 if not isinstance(serial, int):
1277 raise TypeError("serial must be an integer")
1278 if not 0 <= serial < 2**64:
1279 raise ValueError("serial must be between 0 and 2**64")
1280 if self._serial is not None:
1281 raise ValueError("serial already set")
1283 return SSHCertificateBuilder(
1284 _public_key=self._public_key,
1285 _serial=serial,
1286 _type=self._type,
1287 _key_id=self._key_id,
1288 _valid_principals=self._valid_principals,
1289 _valid_for_all_principals=self._valid_for_all_principals,
1290 _valid_before=self._valid_before,
1291 _valid_after=self._valid_after,
1292 _critical_options=self._critical_options,
1293 _extensions=self._extensions,
1294 )
1296 def type(self, type: SSHCertificateType) -> SSHCertificateBuilder:
1297 if not isinstance(type, SSHCertificateType):
1298 raise TypeError("type must be an SSHCertificateType")
1299 if self._type is not None:
1300 raise ValueError("type already set")
1302 return SSHCertificateBuilder(
1303 _public_key=self._public_key,
1304 _serial=self._serial,
1305 _type=type,
1306 _key_id=self._key_id,
1307 _valid_principals=self._valid_principals,
1308 _valid_for_all_principals=self._valid_for_all_principals,
1309 _valid_before=self._valid_before,
1310 _valid_after=self._valid_after,
1311 _critical_options=self._critical_options,
1312 _extensions=self._extensions,
1313 )
1315 def key_id(self, key_id: bytes) -> SSHCertificateBuilder:
1316 if not isinstance(key_id, bytes):
1317 raise TypeError("key_id must be bytes")
1318 if self._key_id is not None:
1319 raise ValueError("key_id already set")
1321 return SSHCertificateBuilder(
1322 _public_key=self._public_key,
1323 _serial=self._serial,
1324 _type=self._type,
1325 _key_id=key_id,
1326 _valid_principals=self._valid_principals,
1327 _valid_for_all_principals=self._valid_for_all_principals,
1328 _valid_before=self._valid_before,
1329 _valid_after=self._valid_after,
1330 _critical_options=self._critical_options,
1331 _extensions=self._extensions,
1332 )
1334 def valid_principals(
1335 self, valid_principals: list[bytes]
1336 ) -> SSHCertificateBuilder:
1337 if self._valid_for_all_principals:
1338 raise ValueError(
1339 "Principals can't be set because the cert is valid "
1340 "for all principals"
1341 )
1342 if (
1343 not all(isinstance(x, bytes) for x in valid_principals)
1344 or not valid_principals
1345 ):
1346 raise TypeError(
1347 "principals must be a list of bytes and can't be empty"
1348 )
1349 if self._valid_principals:
1350 raise ValueError("valid_principals already set")
1352 if len(valid_principals) > _SSHKEY_CERT_MAX_PRINCIPALS:
1353 raise ValueError(
1354 "Reached or exceeded the maximum number of valid_principals"
1355 )
1357 return SSHCertificateBuilder(
1358 _public_key=self._public_key,
1359 _serial=self._serial,
1360 _type=self._type,
1361 _key_id=self._key_id,
1362 _valid_principals=valid_principals,
1363 _valid_for_all_principals=self._valid_for_all_principals,
1364 _valid_before=self._valid_before,
1365 _valid_after=self._valid_after,
1366 _critical_options=self._critical_options,
1367 _extensions=self._extensions,
1368 )
1370 def valid_for_all_principals(self):
1371 if self._valid_principals:
1372 raise ValueError(
1373 "valid_principals already set, can't set "
1374 "valid_for_all_principals"
1375 )
1376 if self._valid_for_all_principals:
1377 raise ValueError("valid_for_all_principals already set")
1379 return SSHCertificateBuilder(
1380 _public_key=self._public_key,
1381 _serial=self._serial,
1382 _type=self._type,
1383 _key_id=self._key_id,
1384 _valid_principals=self._valid_principals,
1385 _valid_for_all_principals=True,
1386 _valid_before=self._valid_before,
1387 _valid_after=self._valid_after,
1388 _critical_options=self._critical_options,
1389 _extensions=self._extensions,
1390 )
1392 def valid_before(self, valid_before: int | float) -> SSHCertificateBuilder:
1393 if not isinstance(valid_before, (int, float)):
1394 raise TypeError("valid_before must be an int or float")
1395 valid_before = int(valid_before)
1396 if valid_before < 0 or valid_before >= 2**64:
1397 raise ValueError("valid_before must [0, 2**64)")
1398 if self._valid_before is not None:
1399 raise ValueError("valid_before already set")
1401 return SSHCertificateBuilder(
1402 _public_key=self._public_key,
1403 _serial=self._serial,
1404 _type=self._type,
1405 _key_id=self._key_id,
1406 _valid_principals=self._valid_principals,
1407 _valid_for_all_principals=self._valid_for_all_principals,
1408 _valid_before=valid_before,
1409 _valid_after=self._valid_after,
1410 _critical_options=self._critical_options,
1411 _extensions=self._extensions,
1412 )
1414 def valid_after(self, valid_after: int | float) -> SSHCertificateBuilder:
1415 if not isinstance(valid_after, (int, float)):
1416 raise TypeError("valid_after must be an int or float")
1417 valid_after = int(valid_after)
1418 if valid_after < 0 or valid_after >= 2**64:
1419 raise ValueError("valid_after must [0, 2**64)")
1420 if self._valid_after is not None:
1421 raise ValueError("valid_after already set")
1423 return SSHCertificateBuilder(
1424 _public_key=self._public_key,
1425 _serial=self._serial,
1426 _type=self._type,
1427 _key_id=self._key_id,
1428 _valid_principals=self._valid_principals,
1429 _valid_for_all_principals=self._valid_for_all_principals,
1430 _valid_before=self._valid_before,
1431 _valid_after=valid_after,
1432 _critical_options=self._critical_options,
1433 _extensions=self._extensions,
1434 )
1436 def add_critical_option(
1437 self, name: bytes, value: bytes
1438 ) -> SSHCertificateBuilder:
1439 if not isinstance(name, bytes) or not isinstance(value, bytes):
1440 raise TypeError("name and value must be bytes")
1441 # This is O(n**2)
1442 if name in [name for name, _ in self._critical_options]:
1443 raise ValueError("Duplicate critical option name")
1445 return SSHCertificateBuilder(
1446 _public_key=self._public_key,
1447 _serial=self._serial,
1448 _type=self._type,
1449 _key_id=self._key_id,
1450 _valid_principals=self._valid_principals,
1451 _valid_for_all_principals=self._valid_for_all_principals,
1452 _valid_before=self._valid_before,
1453 _valid_after=self._valid_after,
1454 _critical_options=[*self._critical_options, (name, value)],
1455 _extensions=self._extensions,
1456 )
1458 def add_extension(
1459 self, name: bytes, value: bytes
1460 ) -> SSHCertificateBuilder:
1461 if not isinstance(name, bytes) or not isinstance(value, bytes):
1462 raise TypeError("name and value must be bytes")
1463 # This is O(n**2)
1464 if name in [name for name, _ in self._extensions]:
1465 raise ValueError("Duplicate extension name")
1467 return SSHCertificateBuilder(
1468 _public_key=self._public_key,
1469 _serial=self._serial,
1470 _type=self._type,
1471 _key_id=self._key_id,
1472 _valid_principals=self._valid_principals,
1473 _valid_for_all_principals=self._valid_for_all_principals,
1474 _valid_before=self._valid_before,
1475 _valid_after=self._valid_after,
1476 _critical_options=self._critical_options,
1477 _extensions=[*self._extensions, (name, value)],
1478 )
1480 def sign(self, private_key: SSHCertPrivateKeyTypes) -> SSHCertificate:
1481 if not isinstance(
1482 private_key,
1483 (
1484 ec.EllipticCurvePrivateKey,
1485 rsa.RSAPrivateKey,
1486 ed25519.Ed25519PrivateKey,
1487 ),
1488 ):
1489 raise TypeError("Unsupported private key type")
1491 if self._public_key is None:
1492 raise ValueError("public_key must be set")
1494 # Not required
1495 serial = 0 if self._serial is None else self._serial
1497 if self._type is None:
1498 raise ValueError("type must be set")
1500 # Not required
1501 key_id = b"" if self._key_id is None else self._key_id
1503 # A zero length list is valid, but means the certificate
1504 # is valid for any principal of the specified type. We require
1505 # the user to explicitly set valid_for_all_principals to get
1506 # that behavior.
1507 if not self._valid_principals and not self._valid_for_all_principals:
1508 raise ValueError(
1509 "valid_principals must be set if valid_for_all_principals "
1510 "is False"
1511 )
1513 if self._valid_before is None:
1514 raise ValueError("valid_before must be set")
1516 if self._valid_after is None:
1517 raise ValueError("valid_after must be set")
1519 if self._valid_after > self._valid_before:
1520 raise ValueError("valid_after must be earlier than valid_before")
1522 # lexically sort our byte strings
1523 self._critical_options.sort(key=lambda x: x[0])
1524 self._extensions.sort(key=lambda x: x[0])
1526 key_type = _get_ssh_key_type(self._public_key)
1527 cert_prefix = key_type + _CERT_SUFFIX
1529 # Marshal the bytes to be signed
1530 nonce = os.urandom(32)
1531 kformat = _lookup_kformat(key_type)
1532 f = _FragList()
1533 f.put_sshstr(cert_prefix)
1534 f.put_sshstr(nonce)
1535 kformat.encode_public(self._public_key, f)
1536 f.put_u64(serial)
1537 f.put_u32(self._type.value)
1538 f.put_sshstr(key_id)
1539 fprincipals = _FragList()
1540 for p in self._valid_principals:
1541 fprincipals.put_sshstr(p)
1542 f.put_sshstr(fprincipals.tobytes())
1543 f.put_u64(self._valid_after)
1544 f.put_u64(self._valid_before)
1545 fcrit = _FragList()
1546 for name, value in self._critical_options:
1547 fcrit.put_sshstr(name)
1548 if len(value) > 0:
1549 foptval = _FragList()
1550 foptval.put_sshstr(value)
1551 fcrit.put_sshstr(foptval.tobytes())
1552 else:
1553 fcrit.put_sshstr(value)
1554 f.put_sshstr(fcrit.tobytes())
1555 fext = _FragList()
1556 for name, value in self._extensions:
1557 fext.put_sshstr(name)
1558 if len(value) > 0:
1559 fextval = _FragList()
1560 fextval.put_sshstr(value)
1561 fext.put_sshstr(fextval.tobytes())
1562 else:
1563 fext.put_sshstr(value)
1564 f.put_sshstr(fext.tobytes())
1565 f.put_sshstr(b"") # RESERVED FIELD
1566 # encode CA public key
1567 ca_type = _get_ssh_key_type(private_key)
1568 caformat = _lookup_kformat(ca_type)
1569 caf = _FragList()
1570 caf.put_sshstr(ca_type)
1571 caformat.encode_public(private_key.public_key(), caf)
1572 f.put_sshstr(caf.tobytes())
1573 # Sigs according to the rules defined for the CA's public key
1574 # (RFC4253 section 6.6 for ssh-rsa, RFC5656 for ECDSA,
1575 # and RFC8032 for Ed25519).
1576 if isinstance(private_key, ed25519.Ed25519PrivateKey):
1577 signature = private_key.sign(f.tobytes())
1578 fsig = _FragList()
1579 fsig.put_sshstr(ca_type)
1580 fsig.put_sshstr(signature)
1581 f.put_sshstr(fsig.tobytes())
1582 elif isinstance(private_key, ec.EllipticCurvePrivateKey):
1583 hash_alg = _get_ec_hash_alg(private_key.curve)
1584 signature = private_key.sign(f.tobytes(), ec.ECDSA(hash_alg))
1585 r, s = asym_utils.decode_dss_signature(signature)
1586 fsig = _FragList()
1587 fsig.put_sshstr(ca_type)
1588 fsigblob = _FragList()
1589 fsigblob.put_mpint(r)
1590 fsigblob.put_mpint(s)
1591 fsig.put_sshstr(fsigblob.tobytes())
1592 f.put_sshstr(fsig.tobytes())
1594 else:
1595 assert isinstance(private_key, rsa.RSAPrivateKey)
1596 # Just like Golang, we're going to use SHA512 for RSA
1597 # https://cs.opensource.google/go/x/crypto/+/refs/tags/
1598 # v0.4.0:ssh/certs.go;l=445
1599 # RFC 8332 defines SHA256 and 512 as options
1600 fsig = _FragList()
1601 fsig.put_sshstr(_SSH_RSA_SHA512)
1602 signature = private_key.sign(
1603 f.tobytes(), padding.PKCS1v15(), hashes.SHA512()
1604 )
1605 fsig.put_sshstr(signature)
1606 f.put_sshstr(fsig.tobytes())
1608 cert_data = binascii.b2a_base64(f.tobytes()).strip()
1609 # load_ssh_public_identity returns a union, but this is
1610 # guaranteed to be an SSHCertificate, so we cast to make
1611 # mypy happy.
1612 return typing.cast(
1613 SSHCertificate,
1614 load_ssh_public_identity(b"".join([cert_prefix, b" ", cert_data])),
1615 )