-vcon
- - - - - - - - - -
-
- class
- Vcon:
-
-
-
-
-
- 24class Vcon: - 25 def __init__(self, vcon_dict={}) -> None: - 26 # deep copy - 27 """ - 28 Initialize a Vcon object from a dictionary. - 29 - 30 :param vcon_dict: a dictionary representing a vCon - 31 :type vcon_dict: dict - 32 """ - 33 self.vcon_dict = json.loads(json.dumps(vcon_dict)) - 34 - 35 @classmethod - 36 def build_from_json(cls, json_string: str) -> Vcon: - 37 """ - 38 Initialize a Vcon object from a JSON string. - 39 - 40 :param json_string: a JSON string representing a vCon - 41 :type json_string: str - 42 :return: a Vcon object - 43 :rtype: Vcon - 44 """ - 45 return cls(json.loads(json_string)) - 46 - 47 @classmethod - 48 def build_new(cls) -> Vcon: - 49 """ - 50 Initialize a Vcon object with default values. - 51 - 52 :return: a Vcon object - 53 :rtype: Vcon - 54 """ - 55 vcon_dict = { - 56 "uuid": cls.uuid8_domain_name("strolid.com"), - 57 "vcon": "0.0.1", - 58 "created_at": datetime.now(timezone.utc).isoformat()[:-3] + "+00:00", - 59 "redacted": {}, - 60 "group": [], - 61 "parties": [], - 62 "dialog": [], - 63 "attachments": [], - 64 "analysis": [], - 65 } - 66 return cls(vcon_dict) - 67 - 68 @property - 69 def tags(self) -> Optional[dict]: - 70 """ - 71 Returns the tags attachment. - 72 - 73 :return: the tags attachment - 74 :rtype: dict or None - 75 """ - 76 return self.find_attachment_by_type("tags") - 77 - 78 def get_tag(self, tag_name) -> Optional[dict]: - 79 """ - 80 Returns the value of a tag by name. - 81 - 82 :param tag_name: the name of the tag - 83 :type tag_name: str - 84 :return: the value of the tag or None if not found - 85 :rtype: str or None - 86 """ - 87 tags_attachment = self.find_attachment_by_type("tags") - 88 if not tags_attachment: - 89 return None - 90 tag = next( - 91 (t for t in tags_attachment["body"] if t.startswith(f"{tag_name}:")), None - 92 ) - 93 if not tag: - 94 return None - 95 tag_value = tag.split(":")[1] - 96 return tag_value - 97 - 98 def add_tag(self, tag_name, tag_value) -> None: - 99 """ -100 Adds a tag to the vCon. -101 -102 :param tag_name: the name of the tag -103 :type tag_name: str -104 :param tag_value: the value of the tag -105 :type tag_value: str -106 :return: None -107 :rtype: None -108 """ -109 tags_attachment = self.find_attachment_by_type("tags") -110 if not tags_attachment: -111 tags_attachment = { -112 "type": "tags", -113 "body": [], -114 "encoding": "json", -115 } -116 self.vcon_dict["attachments"].append(tags_attachment) -117 tags_attachment["body"].append(f"{tag_name}:{tag_value}") -118 -119 def find_attachment_by_type(self, type: str) -> Optional[dict]: -120 """ -121 Finds an attachment by type. -122 -123 :param type: the type of the attachment -124 :type type: str -125 :return: the attachment or None if not found -126 :rtype: dict or None -127 """ -128 return next( -129 (a for a in self.vcon_dict["attachments"] if a["type"] == type), None -130 ) -131 -132 def add_attachment(self, *, body: Union[dict, list, str], type: str, encoding="none") -> None: -133 """ -134 Adds an attachment to the vCon. -135 -136 :param body: the body of the attachment -137 :type body: Union[dict, list, str] -138 :param type: the type of the attachment -139 :type type: str -140 :param encoding: the encoding of the attachment body -141 :type encoding: str -142 :return: None -143 :rtype: None -144 """ -145 if encoding not in ['json', 'none', 'base64url']: -146 raise Exception("Invalid encoding") -147 -148 if encoding == "json": -149 try: -150 json.loads(body) -151 except Exception as e: -152 raise Exception("Invalid JSON body: ", e) -153 -154 if encoding == 'base64url': -155 try: -156 base64.urlsafe_b64decode(body) -157 except Exception as e: -158 raise Exception("Invalid base64url body: ", e) -159 -160 attachment = { -161 "type": type, -162 "body": body, -163 "encoding": encoding, -164 } -165 self.vcon_dict["attachments"].append(attachment) -166 -167 def find_analysis_by_type(self, type) -> Any | None: -168 """ -169 Finds an analysis by type. -170 -171 :param type: the type of the analysis -172 :type type: str -173 :return: the analysis or None if not found -174 :rtype: dict or None -175 """ -176 return next((a for a in self.vcon_dict["analysis"] if a["type"] == type), None) -177 -178 def add_analysis(self, *, type: str, dialog: Union[list, int], vendor: str, body: Union[dict, list, str], encoding="none", extra={}) -> None: -179 """ -180 Adds analysis data to the vCon. -181 -182 :param type: the type of the analysis -183 :type type: str -184 :param dialog: the dialog(s) associated with the analysis -185 :type dialog: Union[list, int] -186 :param vendor: the vendor of the analysis -187 :type vendor: str -188 :param body: the body of the analysis -189 :type body: Union[dict, list, str] -190 :param encoding: the encoding of the body -191 :type encoding: str -192 :param extra: extra key-value pairs to include in the analysis -193 :type extra: dict -194 :return: None -195 :rtype: None -196 """ -197 if encoding not in ['json', 'none', 'base64url']: -198 raise Exception("Invalid encoding") -199 -200 if encoding == "json": -201 try: -202 json.loads(body) -203 except Exception as e: -204 raise Exception("Invalid JSON body: ", e) -205 -206 if encoding == 'base64url': -207 try: -208 base64.urlsafe_b64decode(body) -209 except Exception as e: -210 raise Exception("Invalid base64url body: ", e) -211 -212 analysis = { -213 "type": type, -214 "dialog": dialog, -215 "vendor": vendor, -216 "body": body, -217 "encoding": encoding, -218 **extra, -219 } -220 self.vcon_dict["analysis"].append(analysis) -221 -222 def add_party(self, party: Party) -> None: -223 """ -224 Adds a party to the vCon. -225 -226 :param party: the party to add -227 :type party: Party -228 :return: None -229 :rtype: None -230 """ -231 self.vcon_dict["parties"].append(party.to_dict()) -232 -233 def find_party_index(self, by: str, val: str) -> Optional[int]: -234 """ -235 Find the index of a party in the vCon given a key-value pair. -236 -237 :param by: the key to look for -238 :type by: str -239 :param val: the value to look for -240 :type val: str -241 :return: The index of the party if found, None otherwise -242 :rtype: Optional[int] -243 """ -244 return next( -245 ( -246 ind -247 for ind, party in enumerate(self.vcon_dict["parties"]) -248 if _get(party, by) == val -249 ), -250 None, -251 ) -252 -253 -254 def find_dialog(self, by: str, val: str) -> Optional[Dialog]: -255 """ -256 Find a dialog in the vCon given a key-value pair. Convert the dialog to a Dialog object. -257 -258 :param by: the key to look for -259 :type by: str -260 :param val: the value to look for -261 :type val: str -262 :return: The dialog if found, None otherwise -263 :rtype: Optional[dict] -264 """ -265 dialog = next( -266 ( -267 dialog -268 for dialog in self.vcon_dict["dialog"] -269 if _get(dialog, by) == val -270 ), -271 None, -272 ) -273 if dialog: -274 return Dialog(**dialog) -275 return None -276 -277 def add_dialog(self, dialog: Dialog) -> None: -278 """ -279 Add a dialog to the vCon. -280 -281 :param dialog: the dialog to add -282 :type dialog: dict -283 :return: None -284 :rtype: None -285 """ -286 self.vcon_dict["dialog"].append(dialog.to_dict()) -287 -288 def to_json(self) -> str: -289 """ -290 Serialize the vCon to a JSON string. -291 -292 :return: a JSON string representation of the vCon -293 :rtype: str -294 """ -295 tmp_vcon_dict = copy.copy(self.vcon_dict) -296 return json.dumps(tmp_vcon_dict) -297 -298 def to_dict(self) -> dict: -299 """ -300 Serialize the vCon to a dictionary. -301 -302 :return: a dictionary representation of the vCon -303 :rtype: dict -304 """ -305 return json.loads(self.to_json()) -306 -307 def dumps(self) -> str: -308 """ -309 Alias for `to_json()`. -310 -311 :return: a JSON string representation of the vCon -312 :rtype: str -313 """ -314 return self.to_json() -315 -316 @property -317 def parties(self) -> list[Party]: -318 """ -319 Returns the list of parties. -320 -321 :return: a list of parties -322 :rtype: list[Party] -323 """ -324 return [Party(**party) for party in self.vcon_dict.get("parties", [])] -325 -326 @property -327 def dialog(self) -> list: -328 return self.vcon_dict.get("dialog", []) -329 -330 @property -331 def attachments(self) -> list: -332 return self.vcon_dict.get("attachments", []) -333 -334 @property -335 def analysis(self): -336 return self.vcon_dict.get("analysis", []) -337 -338 @property -339 def uuid(self) -> str: -340 return self.vcon_dict["uuid"] -341 -342 @property -343 def vcon(self) -> str: -344 return self.vcon_dict["vcon"] -345 -346 @property -347 def subject(self) -> Optional[str]: -348 return self.vcon_dict.get("subject") -349 -350 @property -351 def created_at(self): -352 return self.vcon_dict.get("created_at") -353 -354 @property -355 def updated_at(self): -356 return self.vcon_dict.get("updated_at") -357 -358 @property -359 def redacted(self): -360 return self.vcon_dict.get("redacted") -361 -362 @property -363 def appended(self): -364 return self.vcon_dict.get("appended") -365 -366 @property -367 def group(self): -368 return self.vcon_dict.get("group", []) -369 -370 @property -371 def meta(self): -372 return self.vcon_dict.get("meta", {}) -373 -374 @staticmethod -375 def uuid8_domain_name(domain_name: str) -> str: -376 sha1_hasher = hashlib.sha1() -377 sha1_hasher.update(bytes(domain_name, "utf-8")) -378 dn_sha1 = sha1_hasher.digest() -379 -380 hash_upper_64 = dn_sha1[0:8] -381 int64 = int.from_bytes(hash_upper_64, byteorder="big") -382 -383 uuid8_domain = Vcon.uuid8_time(int64) -384 -385 return uuid8_domain -386 -387 @staticmethod -388 def uuid8_time(custom_c_62_bits: int) -> str: -389 global _LAST_V8_TIMESTAMP -390 -391 ns = time.time_ns() -392 if _LAST_V8_TIMESTAMP is not None and ns <= _LAST_V8_TIMESTAMP: -393 ns = _LAST_V8_TIMESTAMP + 1 -394 timestamp_ms, timestamp_ns = divmod(ns, 10**6) -395 subsec = uuid6._subsec_encode(timestamp_ns) -396 -397 subsec_a = subsec >> 8 -398 uuid_int = (timestamp_ms & 0xFFFFFFFFFFFF) << 80 -399 uuid_int |= subsec_a << 64 -400 uuid_int |= custom_c_62_bits -401 -402 uuid_str = str(uuid6.UUID(int=uuid_int, version=7)) -403 assert uuid_str[14] == "7" -404 uuid_str = uuid_str[:14] + "8" + uuid_str[15:] -405 -406 return uuid_str -407 -408 -409 def sign(self, private_key) -> None: -410 """ -411 Sign the vCon using JWS. -412 -413 :param private_key: the private key used for signing -414 :type private_key: Union[rsa.RSAPrivateKey, bytes] -415 :return: None -416 :rtype: None -417 """ -418 """Sign the vCon using JWS.""" -419 payload = self.to_json() -420 jws = JsonWebSignature() -421 protected = { -422 "alg": "RS256", -423 "typ": "JWS" -424 } -425 -426 # Convert private key to PEM format if it's not already -427 if isinstance(private_key, rsa.RSAPrivateKey): -428 pem = private_key.private_bytes( -429 encoding=serialization.Encoding.PEM, -430 format=serialization.PrivateFormat.PKCS8, -431 encryption_algorithm=serialization.NoEncryption() -432 ) -433 else: -434 pem = private_key -435 -436 signed = jws.serialize_compact(protected, payload, pem) -437 signed_str = signed.decode('utf-8') -438 header, payload, signature = signed_str.split('.') -439 -440 self.vcon_dict['signatures'] = [{ -441 "protected": header, -442 "signature": signature -443 }] -444 self.vcon_dict['payload'] = payload -445 -446 def verify(self, public_key) -> bool: -447 """Verify the JWS signature of the vCon. -448 -449 :param public_key: the public key used for verification -450 :type public_key: Union[rsa.RSAPublicKey, bytes] -451 :return: True if the signature is valid, False otherwise -452 :rtype: bool -453 """ -454 """Verify the JWS signature of the vCon.""" -455 if 'signatures' not in self.vcon_dict or 'payload' not in self.vcon_dict: -456 raise ValueError("vCon is not signed") -457 -458 jws = JsonWebSignature() -459 signed_data = f"{self.vcon_dict['signatures'][0]['protected']}.{self.vcon_dict['payload']}.{self.vcon_dict['signatures'][0]['signature']}" -460 -461 # Convert public key to PEM format if it's not already -462 if isinstance(public_key, rsa.RSAPublicKey): -463 pem = public_key.public_bytes( -464 encoding=serialization.Encoding.PEM, -465 format=serialization.PublicFormat.SubjectPublicKeyInfo -466 ) -467 else: -468 pem = public_key -469 -470 try: -471 jws.deserialize_compact(signed_data, pem) -472 return True -473 except BadSignatureError: -474 return False -475 -476 @classmethod -477 def generate_key_pair(cls) -> tuple: -478 """ -479 Generate a new RSA key pair for signing vCons. -480 -481 :return: a tuple containing the private key and public key -482 :rtype: tuple[rSA.RSAPrivateKey, rsa.RSAPublicKey] -483 """ -484 private_key = rsa.generate_private_key( -485 public_exponent=65537, -486 key_size=2048 -487 ) -488 public_key = private_key.public_key() -489 return private_key, public_key -
-
-
-
-
-
-
-
-
-
-
-
- Vcon(vcon_dict={})
-
-
-
-
-
- 25 def __init__(self, vcon_dict={}) -> None: -26 # deep copy -27 """ -28 Initialize a Vcon object from a dictionary. -29 -30 :param vcon_dict: a dictionary representing a vCon -31 :type vcon_dict: dict -32 """ -33 self.vcon_dict = json.loads(json.dumps(vcon_dict)) -
Initialize a Vcon object from a dictionary.
- -Parameters
- --
-
- vcon_dict: a dictionary representing a vCon -
-
-
-
-
-
-
-
-
-
-
- 35 @classmethod -36 def build_from_json(cls, json_string: str) -> Vcon: -37 """ -38 Initialize a Vcon object from a JSON string. -39 -40 :param json_string: a JSON string representing a vCon -41 :type json_string: str -42 :return: a Vcon object -43 :rtype: Vcon -44 """ -45 return cls(json.loads(json_string)) -
Initialize a Vcon object from a JSON string.
- -Parameters
- --
-
- json_string: a JSON string representing a vCon -
Returns
- ---a Vcon object
-
-
-
-
-
-
-
-
-
-
-
-
- 47 @classmethod -48 def build_new(cls) -> Vcon: -49 """ -50 Initialize a Vcon object with default values. -51 -52 :return: a Vcon object -53 :rtype: Vcon -54 """ -55 vcon_dict = { -56 "uuid": cls.uuid8_domain_name("strolid.com"), -57 "vcon": "0.0.1", -58 "created_at": datetime.now(timezone.utc).isoformat()[:-3] + "+00:00", -59 "redacted": {}, -60 "group": [], -61 "parties": [], -62 "dialog": [], -63 "attachments": [], -64 "analysis": [], -65 } -66 return cls(vcon_dict) -
Initialize a Vcon object with default values.
- -Returns
- ---a Vcon object
-
-
-
-
-
-
-
-
-
-
-
- def
- get_tag(self, tag_name) -> Optional[dict]:
-
-
-
-
-
- 78 def get_tag(self, tag_name) -> Optional[dict]: -79 """ -80 Returns the value of a tag by name. -81 -82 :param tag_name: the name of the tag -83 :type tag_name: str -84 :return: the value of the tag or None if not found -85 :rtype: str or None -86 """ -87 tags_attachment = self.find_attachment_by_type("tags") -88 if not tags_attachment: -89 return None -90 tag = next( -91 (t for t in tags_attachment["body"] if t.startswith(f"{tag_name}:")), None -92 ) -93 if not tag: -94 return None -95 tag_value = tag.split(":")[1] -96 return tag_value -
Returns the value of a tag by name.
- -Parameters
- --
-
- tag_name: the name of the tag -
Returns
- ---the value of the tag or None if not found
-
-
-
-
-
-
-
-
-
-
-
- def
- add_tag(self, tag_name, tag_value) -> None:
-
-
-
-
-
- 98 def add_tag(self, tag_name, tag_value) -> None: - 99 """ -100 Adds a tag to the vCon. -101 -102 :param tag_name: the name of the tag -103 :type tag_name: str -104 :param tag_value: the value of the tag -105 :type tag_value: str -106 :return: None -107 :rtype: None -108 """ -109 tags_attachment = self.find_attachment_by_type("tags") -110 if not tags_attachment: -111 tags_attachment = { -112 "type": "tags", -113 "body": [], -114 "encoding": "json", -115 } -116 self.vcon_dict["attachments"].append(tags_attachment) -117 tags_attachment["body"].append(f"{tag_name}:{tag_value}") -
Adds a tag to the vCon.
- -Parameters
- --
-
- tag_name: the name of the tag -
- tag_value: the value of the tag -
Returns
- ---None
-
-
-
-
-
-
-
-
-
-
-
- def
- find_attachment_by_type(self, type: str) -> Optional[dict]:
-
-
-
-
-
- 119 def find_attachment_by_type(self, type: str) -> Optional[dict]: -120 """ -121 Finds an attachment by type. -122 -123 :param type: the type of the attachment -124 :type type: str -125 :return: the attachment or None if not found -126 :rtype: dict or None -127 """ -128 return next( -129 (a for a in self.vcon_dict["attachments"] if a["type"] == type), None -130 ) -
Finds an attachment by type.
- -Parameters
- --
-
- type: the type of the attachment -
Returns
- ---the attachment or None if not found
-
-
-
-
-
-
-
-
-
-
-
- def
- add_attachment( self, *, body: Union[dict, list, str], type: str, encoding='none') -> None:
-
-
-
-
-
- 132 def add_attachment(self, *, body: Union[dict, list, str], type: str, encoding="none") -> None: -133 """ -134 Adds an attachment to the vCon. -135 -136 :param body: the body of the attachment -137 :type body: Union[dict, list, str] -138 :param type: the type of the attachment -139 :type type: str -140 :param encoding: the encoding of the attachment body -141 :type encoding: str -142 :return: None -143 :rtype: None -144 """ -145 if encoding not in ['json', 'none', 'base64url']: -146 raise Exception("Invalid encoding") -147 -148 if encoding == "json": -149 try: -150 json.loads(body) -151 except Exception as e: -152 raise Exception("Invalid JSON body: ", e) -153 -154 if encoding == 'base64url': -155 try: -156 base64.urlsafe_b64decode(body) -157 except Exception as e: -158 raise Exception("Invalid base64url body: ", e) -159 -160 attachment = { -161 "type": type, -162 "body": body, -163 "encoding": encoding, -164 } -165 self.vcon_dict["attachments"].append(attachment) -
Adds an attachment to the vCon.
- -Parameters
- --
-
- body: the body of the attachment -
- type: the type of the attachment -
- encoding: the encoding of the attachment body -
Returns
- ---None
-
-
-
-
-
-
-
-
-
-
-
- def
- find_analysis_by_type(self, type) -> typing.Any | None:
-
-
-
-
-
- 167 def find_analysis_by_type(self, type) -> Any | None: -168 """ -169 Finds an analysis by type. -170 -171 :param type: the type of the analysis -172 :type type: str -173 :return: the analysis or None if not found -174 :rtype: dict or None -175 """ -176 return next((a for a in self.vcon_dict["analysis"] if a["type"] == type), None) -
Finds an analysis by type.
- -Parameters
- --
-
- type: the type of the analysis -
Returns
- ---the analysis or None if not found
-
-
-
-
-
-
-
-
-
-
-
- def
- add_analysis( self, *, type: str, dialog: Union[list, int], vendor: str, body: Union[dict, list, str], encoding='none', extra={}) -> None:
-
-
-
-
-
- 178 def add_analysis(self, *, type: str, dialog: Union[list, int], vendor: str, body: Union[dict, list, str], encoding="none", extra={}) -> None: -179 """ -180 Adds analysis data to the vCon. -181 -182 :param type: the type of the analysis -183 :type type: str -184 :param dialog: the dialog(s) associated with the analysis -185 :type dialog: Union[list, int] -186 :param vendor: the vendor of the analysis -187 :type vendor: str -188 :param body: the body of the analysis -189 :type body: Union[dict, list, str] -190 :param encoding: the encoding of the body -191 :type encoding: str -192 :param extra: extra key-value pairs to include in the analysis -193 :type extra: dict -194 :return: None -195 :rtype: None -196 """ -197 if encoding not in ['json', 'none', 'base64url']: -198 raise Exception("Invalid encoding") -199 -200 if encoding == "json": -201 try: -202 json.loads(body) -203 except Exception as e: -204 raise Exception("Invalid JSON body: ", e) -205 -206 if encoding == 'base64url': -207 try: -208 base64.urlsafe_b64decode(body) -209 except Exception as e: -210 raise Exception("Invalid base64url body: ", e) -211 -212 analysis = { -213 "type": type, -214 "dialog": dialog, -215 "vendor": vendor, -216 "body": body, -217 "encoding": encoding, -218 **extra, -219 } -220 self.vcon_dict["analysis"].append(analysis) -
Adds analysis data to the vCon.
- -Parameters
- --
-
- type: the type of the analysis -
- dialog: the dialog(s) associated with the analysis -
- vendor: the vendor of the analysis -
- body: the body of the analysis -
- encoding: the encoding of the body -
- extra: extra key-value pairs to include in the analysis -
Returns
- ---None
-
-
-
-
-
-
-
-
-
-
-
- 222 def add_party(self, party: Party) -> None: -223 """ -224 Adds a party to the vCon. -225 -226 :param party: the party to add -227 :type party: Party -228 :return: None -229 :rtype: None -230 """ -231 self.vcon_dict["parties"].append(party.to_dict()) -
Adds a party to the vCon.
- -Parameters
- --
-
- party: the party to add -
Returns
- ---None
-
-
-
-
-
-
-
-
-
-
-
- def
- find_party_index(self, by: str, val: str) -> Optional[int]:
-
-
-
-
-
- 233 def find_party_index(self, by: str, val: str) -> Optional[int]: -234 """ -235 Find the index of a party in the vCon given a key-value pair. -236 -237 :param by: the key to look for -238 :type by: str -239 :param val: the value to look for -240 :type val: str -241 :return: The index of the party if found, None otherwise -242 :rtype: Optional[int] -243 """ -244 return next( -245 ( -246 ind -247 for ind, party in enumerate(self.vcon_dict["parties"]) -248 if _get(party, by) == val -249 ), -250 None, -251 ) -
Find the index of a party in the vCon given a key-value pair.
- -Parameters
- --
-
- by: the key to look for -
- val: the value to look for -
Returns
- ---The index of the party if found, None otherwise
-
-
-
-
-
-
-
-
-
-
-
- 254 def find_dialog(self, by: str, val: str) -> Optional[Dialog]: -255 """ -256 Find a dialog in the vCon given a key-value pair. Convert the dialog to a Dialog object. -257 -258 :param by: the key to look for -259 :type by: str -260 :param val: the value to look for -261 :type val: str -262 :return: The dialog if found, None otherwise -263 :rtype: Optional[dict] -264 """ -265 dialog = next( -266 ( -267 dialog -268 for dialog in self.vcon_dict["dialog"] -269 if _get(dialog, by) == val -270 ), -271 None, -272 ) -273 if dialog: -274 return Dialog(**dialog) -275 return None -
Find a dialog in the vCon given a key-value pair. Convert the dialog to a Dialog object.
- -Parameters
- --
-
- by: the key to look for -
- val: the value to look for -
Returns
- ---The dialog if found, None otherwise
-
-
-
-
-
-
-
-
-
-
-
- 277 def add_dialog(self, dialog: Dialog) -> None: -278 """ -279 Add a dialog to the vCon. -280 -281 :param dialog: the dialog to add -282 :type dialog: dict -283 :return: None -284 :rtype: None -285 """ -286 self.vcon_dict["dialog"].append(dialog.to_dict()) -
Add a dialog to the vCon.
- -Parameters
- --
-
- dialog: the dialog to add -
Returns
- ---None
-
-
-
-
-
-
-
-
-
-
-
- def
- to_json(self) -> str:
-
-
-
-
-
- 288 def to_json(self) -> str: -289 """ -290 Serialize the vCon to a JSON string. -291 -292 :return: a JSON string representation of the vCon -293 :rtype: str -294 """ -295 tmp_vcon_dict = copy.copy(self.vcon_dict) -296 return json.dumps(tmp_vcon_dict) -
Serialize the vCon to a JSON string.
- -Returns
- ---a JSON string representation of the vCon
-
-
-
-
-
-
-
-
-
-
-
- def
- to_dict(self) -> dict:
-
-
-
-
-
- 298 def to_dict(self) -> dict: -299 """ -300 Serialize the vCon to a dictionary. -301 -302 :return: a dictionary representation of the vCon -303 :rtype: dict -304 """ -305 return json.loads(self.to_json()) -
Serialize the vCon to a dictionary.
- -Returns
- ---a dictionary representation of the vCon
-
-
-
-
-
- def
- dumps(self) -> str:
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- parties: list[vcon.party.Party]
-
-
-
-
-
- 316 @property -317 def parties(self) -> list[Party]: -318 """ -319 Returns the list of parties. -320 -321 :return: a list of parties -322 :rtype: list[Party] -323 """ -324 return [Party(**party) for party in self.vcon_dict.get("parties", [])] -
Returns the list of parties.
- -Returns
- ---a list of parties
-
-
-
-
-
-
-
-
-
-
-
- @staticmethod
-
- def
- uuid8_domain_name(domain_name: str) -> str:
-
-
-
- 374 @staticmethod -375 def uuid8_domain_name(domain_name: str) -> str: -376 sha1_hasher = hashlib.sha1() -377 sha1_hasher.update(bytes(domain_name, "utf-8")) -378 dn_sha1 = sha1_hasher.digest() -379 -380 hash_upper_64 = dn_sha1[0:8] -381 int64 = int.from_bytes(hash_upper_64, byteorder="big") -382 -383 uuid8_domain = Vcon.uuid8_time(int64) -384 -385 return uuid8_domain -
-
-
-
-
-
-
-
-
-
-
- @staticmethod
-
- def
- uuid8_time(custom_c_62_bits: int) -> str:
-
-
-
- 387 @staticmethod -388 def uuid8_time(custom_c_62_bits: int) -> str: -389 global _LAST_V8_TIMESTAMP -390 -391 ns = time.time_ns() -392 if _LAST_V8_TIMESTAMP is not None and ns <= _LAST_V8_TIMESTAMP: -393 ns = _LAST_V8_TIMESTAMP + 1 -394 timestamp_ms, timestamp_ns = divmod(ns, 10**6) -395 subsec = uuid6._subsec_encode(timestamp_ns) -396 -397 subsec_a = subsec >> 8 -398 uuid_int = (timestamp_ms & 0xFFFFFFFFFFFF) << 80 -399 uuid_int |= subsec_a << 64 -400 uuid_int |= custom_c_62_bits -401 -402 uuid_str = str(uuid6.UUID(int=uuid_int, version=7)) -403 assert uuid_str[14] == "7" -404 uuid_str = uuid_str[:14] + "8" + uuid_str[15:] -405 -406 return uuid_str -
-
-
-
-
-
-
-
-
-
-
- def
- sign(self, private_key) -> None:
-
-
-
-
-
- 409 def sign(self, private_key) -> None: -410 """ -411 Sign the vCon using JWS. -412 -413 :param private_key: the private key used for signing -414 :type private_key: Union[rsa.RSAPrivateKey, bytes] -415 :return: None -416 :rtype: None -417 """ -418 """Sign the vCon using JWS.""" -419 payload = self.to_json() -420 jws = JsonWebSignature() -421 protected = { -422 "alg": "RS256", -423 "typ": "JWS" -424 } -425 -426 # Convert private key to PEM format if it's not already -427 if isinstance(private_key, rsa.RSAPrivateKey): -428 pem = private_key.private_bytes( -429 encoding=serialization.Encoding.PEM, -430 format=serialization.PrivateFormat.PKCS8, -431 encryption_algorithm=serialization.NoEncryption() -432 ) -433 else: -434 pem = private_key -435 -436 signed = jws.serialize_compact(protected, payload, pem) -437 signed_str = signed.decode('utf-8') -438 header, payload, signature = signed_str.split('.') -439 -440 self.vcon_dict['signatures'] = [{ -441 "protected": header, -442 "signature": signature -443 }] -444 self.vcon_dict['payload'] = payload -
Sign the vCon using JWS.
- -Parameters
- --
-
- private_key: the private key used for signing -
Returns
- ---None
-
-
-
-
-
-
-
-
-
-
-
- def
- verify(self, public_key) -> bool:
-
-
-
-
-
- 446 def verify(self, public_key) -> bool: -447 """Verify the JWS signature of the vCon. -448 -449 :param public_key: the public key used for verification -450 :type public_key: Union[rsa.RSAPublicKey, bytes] -451 :return: True if the signature is valid, False otherwise -452 :rtype: bool -453 """ -454 """Verify the JWS signature of the vCon.""" -455 if 'signatures' not in self.vcon_dict or 'payload' not in self.vcon_dict: -456 raise ValueError("vCon is not signed") -457 -458 jws = JsonWebSignature() -459 signed_data = f"{self.vcon_dict['signatures'][0]['protected']}.{self.vcon_dict['payload']}.{self.vcon_dict['signatures'][0]['signature']}" -460 -461 # Convert public key to PEM format if it's not already -462 if isinstance(public_key, rsa.RSAPublicKey): -463 pem = public_key.public_bytes( -464 encoding=serialization.Encoding.PEM, -465 format=serialization.PublicFormat.SubjectPublicKeyInfo -466 ) -467 else: -468 pem = public_key -469 -470 try: -471 jws.deserialize_compact(signed_data, pem) -472 return True -473 except BadSignatureError: -474 return False -
Verify the JWS signature of the vCon.
- -Parameters
- --
-
- public_key: the public key used for verification -
Returns
- ---True if the signature is valid, False otherwise
-
-
-
-
-
-
-
-
-
-
-
-
- @classmethod
-
- def
- generate_key_pair(cls) -> tuple:
-
-
-
- 476 @classmethod -477 def generate_key_pair(cls) -> tuple: -478 """ -479 Generate a new RSA key pair for signing vCons. -480 -481 :return: a tuple containing the private key and public key -482 :rtype: tuple[rSA.RSAPrivateKey, rsa.RSAPublicKey] -483 """ -484 private_key = rsa.generate_private_key( -485 public_exponent=65537, -486 key_size=2048 -487 ) -488 public_key = private_key.public_key() -489 return private_key, public_key -
Generate a new RSA key pair for signing vCons.
- -Returns
- ---a tuple containing the private key and public key
-