main
py 121 lines 3.54 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 from __future__ import annotations
5
6 import hashlib
7 import pickle
8 from _hashlib import HASH, UnsupportedDigestmodError
9 from collections.abc import Callable
10 from typing import Any
11
12 import cbor2
13
14 try:
15 # It is important that this remains an optional dependency.
16 # It would not be allowed in environments with strict security controls,
17 # so it's best not to have it installed when not in use.
18 import xxhash as _xxhash
19
20 if not hasattr(_xxhash, "xxh3_128_digest"):
21 _xxhash = None
22 except ImportError: # pragma: no cover
23 _xxhash = None
24
25
26 def sha256(input: Any) -> bytes:
27 """Hash any picklable Python object using SHA-256.
28
29 The input is serialized using pickle before hashing, which allows
30 arbitrary Python objects to be used. Note that this function does
31 not use a hash seed—if you need one, prepend it explicitly to the input.
32
33 Args:
34 input: Any picklable Python object.
35
36 Returns:
37 Bytes representing the SHA-256 hash of the serialized input.
38
39 """
40 input_bytes = pickle.dumps(input, protocol=pickle.HIGHEST_PROTOCOL)
41 return hashlib.sha256(input_bytes).digest()
42
43
44 def sha256_cbor(input: Any) -> bytes:
45 """Hash objects using CBOR serialization and SHA-256.
46
47 This option is useful for non-Python-dependent serialization and hashing.
48
49 Args:
50 input: Object to be serialized and hashed. Supported types include
51 basic Python types and complex structures like lists, tuples, and
52 dictionaries.
53 Custom classes must implement CBOR serialization methods.
54
55 Returns:
56 Bytes representing the SHA-256 hash of the CBOR serialized input.
57
58 """
59 input_bytes = cbor2.dumps(input, canonical=True)
60 return hashlib.sha256(input_bytes).digest()
61
62
63 def _xxhash_digest(input_bytes: bytes) -> bytes:
64 if _xxhash is None:
65 raise ModuleNotFoundError(
66 "xxhash is required for the 'xxhash' prefix caching hash algorithms. "
67 "Install it via `pip install xxhash`."
68 )
69 return _xxhash.xxh3_128_digest(input_bytes)
70
71
72 def xxhash(input: Any) -> bytes:
73 """Hash picklable objects using xxHash."""
74 input_bytes = pickle.dumps(input, protocol=pickle.HIGHEST_PROTOCOL)
75 return _xxhash_digest(input_bytes)
76
77
78 def xxhash_cbor(input: Any) -> bytes:
79 """Hash objects serialized with CBOR using xxHash."""
80 input_bytes = cbor2.dumps(input, canonical=True)
81 return _xxhash_digest(input_bytes)
82
83
84 def get_hash_fn_by_name(hash_fn_name: str) -> Callable[[Any], bytes]:
85 """Get a hash function by name, or raise an error if the function is not found.
86
87 Args:
88 hash_fn_name: Name of the hash function.
89
90 Returns:
91 A hash function.
92
93 """
94 if hash_fn_name == "sha256":
95 return sha256
96 if hash_fn_name == "sha256_cbor":
97 return sha256_cbor
98 if hash_fn_name == "xxhash":
99 return xxhash
100 if hash_fn_name == "xxhash_cbor":
101 return xxhash_cbor
102
103 raise ValueError(f"Unsupported hash function: {hash_fn_name}")
104
105
106 def safe_hash(data: bytes, usedforsecurity: bool = True) -> HASH:
107 """Hash for configs, defaulting to md5 but falling back to sha256
108 in FIPS constrained environments.
109
110 Args:
111 data: bytes
112 usedforsecurity: Whether the hash is used for security purposes
113
114 Returns:
115 Hash object
116
117 """
118 try:
119 return hashlib.md5(data, usedforsecurity=usedforsecurity)
120 except (UnsupportedDigestmodError, ValueError):
121 return hashlib.sha256(data)