distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""DSA — Signature type registry, key generator, and sign/verify dispatcher.
2
3Ported from net.i2p.crypto.DSAEngine, net.i2p.crypto.SigType,
4and net.i2p.crypto.KeyGenerator.
5
6Dispatches to the correct algorithm based on SigType:
7- DSA_SHA1 (code 0): Legacy DSA with SHA-1 [keygen only, sign/verify not implemented]
8- ECDSA_SHA256_P256 (code 1): ECDSA with NIST P-256
9- ECDSA_SHA384_P384 (code 2): ECDSA with NIST P-384
10- ECDSA_SHA512_P521 (code 3): ECDSA with NIST P-521
11- EdDSA_SHA512_Ed25519 (code 7): Ed25519
12- RedDSA_SHA512_Ed25519 (code 11): RedDSA (same key format as Ed25519)
13"""
14
15from __future__ import annotations
16
17import enum
18
19from cryptography.hazmat.primitives.asymmetric.ec import (
20 ECDSA,
21 SECP256R1,
22 SECP384R1,
23 SECP521R1,
24 EllipticCurvePrivateKey,
25 EllipticCurvePublicKey,
26 generate_private_key,
27)
28from cryptography.hazmat.primitives.asymmetric.ed25519 import (
29 Ed25519PrivateKey,
30 Ed25519PublicKey,
31)
32from cryptography.hazmat.primitives.asymmetric.utils import (
33 decode_dss_signature,
34 encode_dss_signature,
35)
36from cryptography.hazmat.primitives.hashes import SHA256, SHA384, SHA512
37from cryptography.hazmat.primitives.serialization import (
38 Encoding,
39 PublicFormat,
40 PrivateFormat,
41 NoEncryption,
42)
43from cryptography.exceptions import InvalidSignature
44
45
46class SigType(enum.Enum):
47 """I2P signature type registry.
48
49 Each member defines the algorithm, key sizes, signature size,
50 and hash algorithm for one signature type code.
51 """
52
53 # code pubkey_len privkey_len sig_len hash_algo
54 DSA_SHA1 = (0, 128, 20, 40, "sha1")
55 ECDSA_SHA256_P256 = (1, 64, 32, 64, "sha256")
56 ECDSA_SHA384_P384 = (2, 96, 48, 96, "sha384")
57 ECDSA_SHA512_P521 = (3, 132, 66, 132, "sha512")
58 EdDSA_SHA512_Ed25519 = (7, 32, 32, 64, "sha512")
59 RedDSA_SHA512_Ed25519 = (11, 32, 32, 64, "sha512")
60
61 def __init__(self, code: int, pubkey_len: int, privkey_len: int,
62 sig_len: int, hash_algo: str) -> None:
63 self._code = code
64 self._pubkey_len = pubkey_len
65 self._privkey_len = privkey_len
66 self._sig_len = sig_len
67 self._hash_algo = hash_algo
68
69 @property
70 def code(self) -> int:
71 return self._code
72
73 @property
74 def pubkey_len(self) -> int:
75 return self._pubkey_len
76
77 @property
78 def privkey_len(self) -> int:
79 return self._privkey_len
80
81 @property
82 def sig_len(self) -> int:
83 return self._sig_len
84
85 @property
86 def hash_algo(self) -> str:
87 return self._hash_algo
88
89 @classmethod
90 def by_code(cls, code: int) -> SigType | None:
91 """Look up a SigType by its integer code, or None."""
92 for st in cls:
93 if st.code == code:
94 return st
95 return None
96
97
98_ECDSA_CURVES = {
99 SigType.ECDSA_SHA256_P256: (SECP256R1, SHA256, 32),
100 SigType.ECDSA_SHA384_P384: (SECP384R1, SHA384, 48),
101 SigType.ECDSA_SHA512_P521: (SECP521R1, SHA512, 66),
102}
103
104
105def _ecdsa_sig_to_fixed(der_sig: bytes, component_len: int) -> bytes:
106 """Convert DER-encoded ECDSA signature to fixed-length (r||s)."""
107 r, s = decode_dss_signature(der_sig)
108 return r.to_bytes(component_len, "big") + s.to_bytes(component_len, "big")
109
110
111def _fixed_to_ecdsa_sig(fixed: bytes, component_len: int) -> bytes:
112 """Convert fixed-length (r||s) to DER-encoded ECDSA signature."""
113 r = int.from_bytes(fixed[:component_len], "big")
114 s = int.from_bytes(fixed[component_len:2 * component_len], "big")
115 return encode_dss_signature(r, s)
116
117
118class KeyGenerator:
119 """Generate key pairs for any SigType."""
120
121 @staticmethod
122 def generate(sig_type: SigType) -> tuple[bytes, bytes]:
123 """Generate a key pair.
124
125 Args:
126 sig_type: the signature type
127
128 Returns:
129 (public_key_bytes, private_key_bytes)
130 """
131 if sig_type in (SigType.EdDSA_SHA512_Ed25519,
132 SigType.RedDSA_SHA512_Ed25519):
133 ed_key = Ed25519PrivateKey.generate()
134 pub = ed_key.public_key().public_bytes_raw()
135 priv = ed_key.private_bytes_raw()
136 return pub, priv
137
138 if sig_type in _ECDSA_CURVES:
139 curve_cls, _, _ = _ECDSA_CURVES[sig_type]
140 ec_key = generate_private_key(curve_cls())
141 # Raw public key: uncompressed point without 0x04 prefix
142 pub_uncompressed = ec_key.public_key().public_bytes(
143 Encoding.X962, PublicFormat.UncompressedPoint
144 )
145 pub = pub_uncompressed[1:] # strip 0x04 prefix
146 # Extract raw private scalar
147 priv_num = ec_key.private_numbers().private_value
148 priv_bytes = priv_num.to_bytes(sig_type.privkey_len, "big")
149 return pub, priv_bytes
150
151 if sig_type == SigType.DSA_SHA1:
152 from cryptography.hazmat.primitives.asymmetric.dsa import (
153 generate_parameters, generate_private_key as dsa_gen_key,
154 )
155 params = generate_parameters(key_size=1024)
156 dsa_key = params.generate_private_key()
157 pub_num = dsa_key.public_key().public_numbers()
158 dsa_priv_num = dsa_key.private_numbers()
159 pub = pub_num.y.to_bytes(128, "big")
160 priv = dsa_priv_num.x.to_bytes(20, "big")
161 return pub, priv
162
163 raise ValueError(f"Unsupported SigType: {sig_type}")
164
165
166class DSAEngine:
167 """Signature dispatcher — routes sign/verify to the correct algorithm."""
168
169 @staticmethod
170 def sign(data: bytes, private_key: bytes, sig_type: SigType) -> bytes:
171 """Sign data using the specified signature type.
172
173 Args:
174 data: message to sign
175 private_key: raw private key bytes
176 sig_type: signature algorithm
177
178 Returns:
179 Fixed-length signature (sig_type.sig_len bytes)
180 """
181 if sig_type in (SigType.EdDSA_SHA512_Ed25519,
182 SigType.RedDSA_SHA512_Ed25519):
183 ed_key = Ed25519PrivateKey.from_private_bytes(private_key[:32])
184 return ed_key.sign(data)
185
186 if sig_type in _ECDSA_CURVES:
187 curve_cls, hash_cls, component_len = _ECDSA_CURVES[sig_type]
188 # Reconstruct private key from raw scalar
189 from cryptography.hazmat.primitives.asymmetric.ec import (
190 derive_private_key,
191 )
192 d = int.from_bytes(private_key[:sig_type.privkey_len], "big")
193 ec_key = derive_private_key(d, curve_cls())
194 der_sig = ec_key.sign(data, ECDSA(hash_cls()))
195 return _ecdsa_sig_to_fixed(der_sig, component_len)
196
197 raise ValueError(f"sign not implemented for {sig_type}")
198
199 @staticmethod
200 def verify(data: bytes, signature: bytes, public_key: bytes,
201 sig_type: SigType) -> bool:
202 """Verify a signature.
203
204 Args:
205 data: message that was signed
206 signature: fixed-length signature bytes
207 public_key: raw public key bytes
208 sig_type: signature algorithm
209
210 Returns:
211 True if signature is valid
212 """
213 try:
214 if sig_type in (SigType.EdDSA_SHA512_Ed25519,
215 SigType.RedDSA_SHA512_Ed25519):
216 ed_pub = Ed25519PublicKey.from_public_bytes(public_key[:32])
217 ed_pub.verify(signature, data)
218 return True
219
220 if sig_type in _ECDSA_CURVES:
221 curve_cls, hash_cls, component_len = _ECDSA_CURVES[sig_type]
222 # Reconstruct public key from raw x||y
223 point = b"\x04" + public_key[:sig_type.pubkey_len]
224 ec_pub = EllipticCurvePublicKey.from_encoded_point(
225 curve_cls(), point
226 )
227 der_sig = _fixed_to_ecdsa_sig(signature, component_len)
228 ec_pub.verify(der_sig, data, ECDSA(hash_cls()))
229 return True
230
231 except (InvalidSignature, ValueError, OverflowError):
232 return False
233
234 raise ValueError(f"verify not implemented for {sig_type}")