1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
| from cryptography.hazmat.primitives import hashes
import secrets
class DistributedKeyEscrow:
def __init__(self, threshold, total_custodians):
self.threshold = threshold
self.total_custodians = total_custodians
def fragment_key(self, private_key):
"""Fragment key into multiple parts"""
# Serialize private key
private_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
# Convert to bytes
key_bytes = private_pem
# Generate random coefficients
coefficients = [secrets.randbits(256) for _ in range(self.threshold - 1)]
coefficients.insert(0, int.from_bytes(key_bytes, 'big'))
# Generate fragments
fragments = []
for i in range(1, self.total_custodians + 1):
fragment = 0
for j, coeff in enumerate(coefficients):
fragment += coeff * (i ** j)
fragments.append((i, fragment))
return fragments
def reconstruct_key(self, fragments):
"""Reconstruct key from fragments"""
if len(fragments) < self.threshold:
raise ValueError("Insufficient fragments")
# Use Lagrange interpolation
x_values = [f[0] for f in fragments[:self.threshold]]
y_values = [f[1] for f in fragments[:self.threshold]]
# Reconstruct key
key_value = 0
for i in range(self.threshold):
term = y_values[i]
for j in range(self.threshold):
if i != j:
term = term * (-x_values[j]) // (x_values[i] - x_values[j])
key_value += term
# Convert back to bytes
key_bytes = key_value.to_bytes((key_value.bit_length() + 7) // 8, 'big')
# Load private key
private_key = serialization.load_pem_private_key(
key_bytes,
password=None
)
return private_key
# Usage example
distributed_escrow = DistributedKeyEscrow(threshold=3, total_custodians=5)
fragments = distributed_escrow.fragment_key(private_key)
reconstructed = distributed_escrow.reconstruct_key(fragments[:3])
|