Coverage for gcid/gcid.py: 78%
146 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-07-28 08:43 -0700
1"""Cryptographic, location aware ID conversions"""
3import hashlib
4import hmac
5import logging
6from enum import Enum
8import base58
9from cryptography.hazmat.backends import default_backend
10from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
11from pydantic import BaseModel
13from gcid.config import Config
15_config = Config()
17DIGEST_SIZE = 16
19# ID encoding constants
20_HMAC_BYTES = 4
21_SEQ_BYTES = 8
22_PREFIX_BYTES = 8
23_ENCRYPTED_BYTES = _PREFIX_BYTES + _SEQ_BYTES
25_ENC_KEY: bytes = _config.gcid_enc_key.encode("utf-8")
26_HMAC_KEY: bytes = _config.gcid_hmac_key.encode("utf-8")
27_PERSON: bytes = b"id"
29# Generate a 16-byte Initialization Vector (IV)
30_IV = b"\00" * 16
32# ID metadata
33_VERSION = b"\01"
34_LOCATION_PARTITION = b"\00\00\00\00\00\00\00"
35_API_ID_PREFIX = _VERSION + _LOCATION_PARTITION
37# Create a Cipher object using AES algorithm in CBC mode
38cipher = Cipher(algorithms.AES(_ENC_KEY), modes.CBC(_IV), backend=default_backend())
40log = logging.getLogger(__name__)
43class IdType(Enum):
44 """Type enum for ID prefixes"""
46 PROFILE = "prf"
47 ORG = "org"
48 ASSET = "asset"
49 FILE = "file"
50 EVENT = "evt"
51 TOPO = "topo"
52 JOBDEF = "jobdef"
53 JOB = "job"
54 JOBRESULT = "jobres"
55 READER = "read"
56 TAG = "tag"
59class ApiError(Exception):
60 def __init__(self, message):
61 super().__init__(message)
64class IdError(Exception):
65 def __init__(self, id_str: str, reason: str):
66 super().__init__(f"API ID {id_str} is not valid: {reason}")
69class SequenceError(ApiError):
70 def __init__(self, seq_num: int, reason: str):
71 super().__init__(f"sequence number {seq_num} is not valid: {reason}")
74id_rev_map = {i.value: i for i in list(IdType)}
77class DbSeq(BaseModel):
78 id_type: IdType
79 prefix: bytes
80 seq: int
83def seq_to_id(api_type: IdType, seq: int | None) -> str:
84 """Given a 64bit integer, return an encrypted version with a fixed HMAC"""
85 if seq is None:
86 raise SequenceError(0, "sequence is none")
88 if type(seq) is not int: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 raise SequenceError(seq, f"invalid type {type(seq)}")
91 # encrypt first 16 bytes (note this must be done in 128bit blocks or else padded)
92 seq_bytes = seq.to_bytes(8, "big")
93 e = cipher.encryptor()
94 encrypted = e.update(_API_ID_PREFIX + seq_bytes)
95 if len(encrypted) != _ENCRYPTED_BYTES: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true
96 raise SequenceError(seq, f"invalid encrypted length {len(encrypted)}")
98 h = hashlib.blake2b(digest_size=DIGEST_SIZE, key=_HMAC_KEY, person=_PERSON)
99 h.update(encrypted)
100 digest = h.digest()
101 combined = encrypted + digest[0:_HMAC_BYTES]
102 encoded = base58.b58encode(combined)
104 return "".join((api_type.value, "_", encoded.decode("utf-8")))
107def id_type(api_id: str) -> IdType:
108 """Return the ID type from the API ID"""
109 if type(api_id) is not str: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 raise IdError(api_id, f"invalid api id type {type(api_id)}")
112 parts = api_id.split("_")
113 if len(parts) != 2: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true
114 raise IdError(api_id, f"invalid api id format {api_id}")
116 type_str, api_id = parts
117 return id_rev_map[type_str]
120def id_to_seq(api_id: str, api_id_type: IdType) -> int:
121 """Validate and decode an encrypted serial number"""
122 if type(api_id) is not str: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 raise IdError(api_id, f"IDs must be of string type: {type(api_id)}")
125 parts = api_id.split("_")
126 if len(parts) != 2:
127 raise IdError(api_id, f"ID has invalid format: {api_id}")
129 type_str, api_id = parts
130 if api_id_type != id_rev_map.get(type_str):
131 raise IdError(api_id, f"ID has invalid type: {type_str}")
133 # Base58 decode the obfuscated serial number
134 combined = base58.b58decode(api_id)
136 # Extract the encrypted serial number and HMAC
137 encrypted: bytes = combined[0:_ENCRYPTED_BYTES]
138 serial_hmac = combined[_ENCRYPTED_BYTES:]
139 if len(serial_hmac) != _HMAC_BYTES: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 raise IdError(api_id, "HMAC byte count mismatch")
142 # Verify the HMAC
143 h = hashlib.blake2b(digest_size=DIGEST_SIZE, key=_HMAC_KEY, person=_PERSON)
144 h.update(encrypted)
145 digest = h.digest()
146 if not hmac.compare_digest(serial_hmac, digest[0:_HMAC_BYTES]):
147 raise IdError(api_id, "Invalid HMAC - data may have been tampered with")
149 # Decrypt the serial number
150 d = cipher.decryptor()
151 decrypted_data = d.update(encrypted)
152 prefix = decrypted_data[0:_PREFIX_BYTES]
153 seq = decrypted_data[_PREFIX_BYTES:]
154 if len(seq) != _SEQ_BYTES: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 raise IdError(api_id, f"Invalid sequence bytes {len(seq)}")
157 if prefix != _API_ID_PREFIX: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 raise IdError(api_id, f"ID has invalid prefix: {prefix}")
160 return int.from_bytes(seq, "big")
163def asset_seq_to_id(asset_seq: int) -> str:
164 """Convert asset seq to ID"""
165 return seq_to_id(IdType.ASSET, asset_seq)
168def asset_id_to_seq(asset_id: str) -> int:
169 """Convert asset ID to encrypted seq"""
170 return id_to_seq(asset_id, IdType.ASSET)
173def profile_seq_to_id(profile_seq: int) -> str:
174 """Convert asset seq to ID"""
175 return seq_to_id(IdType.PROFILE, profile_seq)
178def profile_id_to_seq(profile_id: str) -> int:
179 """Convert asset ID to encrypted seq"""
180 return id_to_seq(profile_id, IdType.PROFILE)
183def org_seq_to_id(org_seq: int) -> str:
184 """Convert asset seq to ID"""
185 return seq_to_id(IdType.ORG, org_seq)
188def org_id_to_seq(org_id: str) -> int | None:
189 """Convert asset ID to encrypted seq"""
190 return id_to_seq(org_id, IdType.ORG)
193def file_seq_to_id(file_seq: int) -> str:
194 """Convert asset seq to ID"""
195 return seq_to_id(IdType.FILE, file_seq)
198def file_id_to_seq(file_id: str) -> int:
199 """Convert asset ID to encrypted seq"""
200 return id_to_seq(file_id, IdType.FILE)
203def event_seq_to_id(file_seq: int) -> str:
204 """Convert asset seq to ID"""
205 return seq_to_id(IdType.EVENT, file_seq)
208def event_id_to_seq(file_id: str) -> int:
209 """Convert asset ID to encrypted seq"""
210 return id_to_seq(file_id, IdType.EVENT)
213def topo_seq_to_id(topo_seq: int) -> str:
214 """Convert asset seq to ID"""
215 return seq_to_id(IdType.TOPO, topo_seq)
218def topo_id_to_seq(topo_id: str) -> int:
219 """Convert asset ID to encrypted seq"""
220 return id_to_seq(topo_id, IdType.TOPO)
223def job_def_seq_to_id(job_def_seq: int) -> str:
224 """Convert job def seq to ID"""
225 return seq_to_id(IdType.JOBDEF, job_def_seq)
228def job_def_id_to_seq(job_def_id: str) -> int:
229 """Convert job def ID to encrypted seq"""
230 return id_to_seq(job_def_id, IdType.JOBDEF)
233def job_seq_to_id(job_seq: int) -> str:
234 """Convert job seq to ID"""
235 return seq_to_id(IdType.JOB, job_seq)
238def job_id_to_seq(job_id: str) -> int:
239 """Convert job ID to encrypted seq"""
240 return id_to_seq(job_id, IdType.JOB)
243def job_result_seq_to_id(job_result_seq: int) -> str:
244 """Convert job result seq to ID"""
245 return seq_to_id(IdType.JOBRESULT, job_result_seq)
248def job_result_id_to_seq(job_result_id: str) -> int:
249 """Convert job result ID to encrypted seq"""
250 return id_to_seq(job_result_id, IdType.JOBRESULT)
253def reader_seq_to_id(reader_seq: int) -> str:
254 """Convert reader seq to encrypted ID"""
255 return seq_to_id(IdType.READER, reader_seq)
258def reader_id_to_seq(reader_id: str) -> int:
259 """Convert reader ID to database seq"""
260 return id_to_seq(reader_id, IdType.READER)
263def tag_seq_to_id(tag_seq: int) -> str:
264 """Convert tag seq to encrypted ID"""
265 return seq_to_id(IdType.TAG, tag_seq)
268def tag_id_to_seq(tag_id: str) -> int:
269 """Convert tag ID to database seq"""
270 return id_to_seq(tag_id, IdType.TAG)