Coverage for presidio_analyzer / predefined_recognizers / generic / crypto_recognizer.py: 91%
78 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 09:03 +0000
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 09:03 +0000
1from hashlib import sha256
2from typing import List, Optional
4from presidio_analyzer import Pattern, PatternRecognizer
6# This class includes references to addresses validation algorithms.
7# The original implementation of the P2PKH and P2SH address validation
8# algorithm can be found at:
9# http://rosettacode.org/wiki/Bitcoin/address_validation#Python
10# The Bech32/Bech32m address validation algorithm by Pieter Wuille can be found
11# at: https://github.com/sipa/bech32/blob/master/ref/python/segwit_addr.py
13# Constants for encoding types
14BECH32 = 1
15BECH32M = 2
17CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
18BECH32M_CONST = 0x2BC830A3
21class CryptoRecognizer(PatternRecognizer):
22 """Recognize common crypto account numbers using regex + checksum.
24 :param patterns: List of patterns to be used by this recognizer
25 :param context: List of context words to increase confidence in detection
26 :param supported_language: Language this recognizer supports
27 :param supported_entity: The entity this recognizer can detect
28 """
30 PATTERNS = [
31 Pattern("Crypto (Medium)", r"(bc1|[13])[a-zA-HJ-NP-Z0-9]{25,59}", 0.5),
32 ]
34 CONTEXT = ["wallet", "btc", "bitcoin", "crypto"]
36 def __init__(
37 self,
38 patterns: Optional[List[Pattern]] = None,
39 context: Optional[List[str]] = None,
40 supported_language: str = "en",
41 supported_entity: str = "CRYPTO",
42 name: Optional[str] = None,
43 ):
44 patterns = patterns if patterns else self.PATTERNS
45 context = context if context else self.CONTEXT
46 super().__init__(
47 supported_entity=supported_entity,
48 patterns=patterns,
49 context=context,
50 supported_language=supported_language,
51 name=name,
52 )
54 def validate_result(self, pattern_text: str) -> bool:
55 """Validate the Bitcoin address using checksum.
57 :param pattern_text: The cryptocurrency address to validate.
58 :return: True if the address is valid according to its respective
59 format, False otherwise.
60 """
61 if pattern_text.startswith("1") or pattern_text.startswith("3"):
62 # P2PKH or P2SH address validation
63 try:
64 bcbytes = self.__decode_base58(str.encode(pattern_text))
65 checksum = sha256(sha256(bcbytes[:-4]).digest()).digest()[:4]
66 return bcbytes[-4:] == checksum
67 except ValueError:
68 return False
69 elif pattern_text.startswith("bc1"):
70 # Bech32 or Bech32m address validation
71 if CryptoRecognizer.validate_bech32_address(pattern_text)[0]:
72 return True
73 return False
75 @staticmethod
76 def __decode_base58(bc: bytes) -> bytes:
77 digits58 = b"123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
78 origlen = len(bc)
79 bc = bc.lstrip(digits58[0:1])
81 n = 0
82 for char in bc:
83 n = n * 58 + digits58.index(char)
84 return n.to_bytes(origlen - len(bc) + (n.bit_length() + 7) // 8, "big")
86 @staticmethod
87 def bech32_polymod(values):
88 """Compute the Bech32 checksum."""
89 generator = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3]
90 chk = 1
91 for value in values:
92 top = chk >> 25
93 chk = (chk & 0x1FFFFFF) << 5 ^ value
94 for i in range(5):
95 chk ^= generator[i] if ((top >> i) & 1) else 0
96 return chk
98 @staticmethod
99 def bech32_hrp_expand(hrp):
100 """Expand the HRP into values for checksum computation."""
101 return [ord(x) >> 5 for x in hrp] + [0] + [ord(x) & 31 for x in hrp]
103 @staticmethod
104 def bech32_verify_checksum(hrp, data):
105 """Verify a checksum given HRP and converted data characters."""
106 const = CryptoRecognizer.bech32_polymod(
107 CryptoRecognizer.bech32_hrp_expand(hrp) + data
108 )
109 if const == 1:
110 return BECH32
111 if const == BECH32M_CONST:
112 return BECH32M
113 return None
115 @staticmethod
116 def bech32_decode(bech):
117 """Validate a Bech32/Bech32m string, and determine HRP and data."""
118 if (any(ord(x) < 33 or ord(x) > 126 for x in bech)) or (
119 bech.lower() != bech and bech.upper() != bech
120 ):
121 return (None, None, None)
122 bech = bech.lower()
123 pos = bech.rfind("1")
124 if pos < 1 or pos + 7 > len(bech) or len(bech) > 90:
125 return (None, None, None)
126 if not all(x in CHARSET for x in bech[pos + 1 :]):
127 return (None, None, None)
128 hrp = bech[:pos]
129 data = [CHARSET.find(x) for x in bech[pos + 1 :]]
130 spec = CryptoRecognizer.bech32_verify_checksum(hrp, data)
131 if spec is None:
132 return (None, None, None)
133 return (hrp, data[:-6], spec)
135 @staticmethod
136 def validate_bech32_address(address):
137 """Validate a Bech32 or Bech32m address."""
138 hrp, data, spec = CryptoRecognizer.bech32_decode(address)
139 if hrp is not None and data is not None:
140 return True, spec
141 return False, None