1"""Dead Simple Signing Envelope"""
2
3from __future__ import annotations
4
5import logging
6from typing import Any
7
8from securesystemslib import exceptions
9from securesystemslib._internal.utils import b64dec, b64enc, make_hashable
10from securesystemslib.signer import Key, Signature, Signer
11
12logger = logging.getLogger(__name__)
13
14
15class Envelope:
16 """DSSE Envelope to provide interface for signing arbitrary data.
17
18 Attributes:
19 payload: Arbitrary byte sequence of serialized body.
20 payload_type: string that identifies how to interpret payload.
21 signatures: dict of Signature key id and Signatures.
22
23 """
24
25 def __init__(
26 self,
27 payload: bytes,
28 payload_type: str,
29 signatures: dict[str, Signature],
30 ):
31 self.payload = payload
32 self.payload_type = payload_type
33 self.signatures = signatures
34
35 def __eq__(self, other: Any) -> bool:
36 if not isinstance(other, Envelope):
37 return False
38
39 return (
40 self.payload == other.payload
41 and self.payload_type == other.payload_type
42 and self.signatures == other.signatures
43 )
44
45 def __hash__(self) -> int:
46 return hash(
47 (
48 self.payload,
49 self.payload_type,
50 make_hashable(self.signatures),
51 )
52 )
53
54 @classmethod
55 def from_dict(cls, data: dict) -> Envelope:
56 """Creates a DSSE Envelope from its JSON/dict representation.
57
58 Arguments:
59 data: A dict containing a valid payload, payloadType and signatures
60
61 Raises:
62 KeyError: If any of the "payload", "payloadType" and "signatures"
63 fields are missing from the "data".
64
65 FormatError: If signature in "signatures" is incorrect.
66
67 Returns:
68 A "Envelope" instance.
69 """
70
71 payload = b64dec(data["payload"])
72 payload_type = data["payloadType"]
73
74 signatures = {}
75 for signature in data["signatures"]:
76 signature["sig"] = b64dec(signature["sig"]).hex()
77 sig = Signature.from_dict(signature)
78 if sig.keyid in signatures:
79 raise ValueError(f"Multiple signatures found for keyid {sig.keyid}")
80 signatures[sig.keyid] = sig
81 return cls(payload, payload_type, signatures)
82
83 def to_dict(self) -> dict:
84 """Returns the JSON-serializable dictionary representation of self."""
85
86 signatures = []
87 for signature in self.signatures.values():
88 sig_dict = signature.to_dict()
89 sig_dict["sig"] = b64enc(bytes.fromhex(sig_dict["sig"]))
90 signatures.append(sig_dict)
91
92 return {
93 "payload": b64enc(self.payload),
94 "payloadType": self.payload_type,
95 "signatures": signatures,
96 }
97
98 def pae(self) -> bytes:
99 """Pre-Auth-Encoding byte sequence of self."""
100
101 return b"DSSEv1 %d %b %d %b" % (
102 len(self.payload_type),
103 self.payload_type.encode("utf-8"),
104 len(self.payload),
105 self.payload,
106 )
107
108 def sign(self, signer: Signer) -> Signature:
109 """Sign the payload and create the signature.
110
111 Arguments:
112 signer: A "Signer" class instance.
113
114 Returns:
115 A "Signature" instance.
116 """
117
118 signature = signer.sign(self.pae())
119 self.signatures[signature.keyid] = signature
120
121 return signature
122
123 def verify(self, keys: list[Key], threshold: int) -> dict[str, Key]:
124 """Verify the payload with the provided Keys.
125
126 Arguments:
127 keys: A list of public keys to verify the signatures.
128 threshold: Number of signatures needed to pass the verification.
129
130 Raises:
131 ValueError: If "threshold" is not valid.
132 VerificationError: If the enclosed signatures do not pass the
133 verification.
134
135 Note:
136 Mandating keyid in signatures and matching them with keyid of Key
137 in order to consider them for verification, is not DSSE spec
138 compliant (Issue #416).
139
140 Returns:
141 A dict of the threshold of unique public keys that verified a
142 signature.
143 """
144
145 accepted_keys = {}
146 pae = self.pae()
147
148 # checks for threshold value.
149 if threshold <= 0:
150 raise ValueError("Threshold must be greater than 0")
151
152 if len(keys) < threshold:
153 raise ValueError("Number of keys can't be less than threshold")
154
155 for signature in self.signatures.values():
156 for key in keys:
157 # If Signature keyid doesn't match with Key, skip.
158 if not key.keyid == signature.keyid:
159 continue
160
161 # If a key verifies the signature, we exit and use the result.
162 try:
163 key.verify_signature(signature, pae)
164 accepted_keys[key.keyid] = key
165 break
166 except exceptions.UnverifiedSignatureError:
167 continue
168
169 # Break, if amount of accepted_keys are more than threshold.
170 if len(accepted_keys) >= threshold:
171 break
172
173 if threshold > len(accepted_keys):
174 raise exceptions.VerificationError(
175 "Accepted signatures do not match threshold,"
176 f" Found: {len(accepted_keys)}, Expected {threshold}"
177 )
178
179 return accepted_keys