Files
fn_registry/python/functions/cybersecurity/extract_crypto_wallets.py
T
egutierrez cce7764510 feat(cybersecurity): 8 IoC regex extractors + extract_iocs pipeline puro
Extractores nuevos en python/functions/cybersecurity/:
- extract_ip_addresses (IPv4 + IPv6 con validacion ipaddress)
- extract_emails (RFC 5322 simplificado)
- extract_domains (FQDNs con TLD valido, lista estatica)
- extract_file_hashes (MD5/SHA1/SHA256/SHA512, algoritmo por longitud)
- extract_crypto_wallets (BTC legacy + bech32, ETH 0x+40hex)
- extract_cve_ids (CVE-YYYY-NNNN+)
- extract_mac_addresses (xx:xx:xx + xx-xx-xx, separador uniforme)
- extract_phone_numbers (E.164 + ES local 9 digitos)

Pipeline:
- extract_iocs corre todos, deduplica spans contenidos. Mantiene
  purity:pure (kind:function con uses_functions no vacio) porque la
  regla del registry exige que los pipelines sean impuros.

Todas devuelven list[dict] con value/start/end/type para que el
caller (issues 0038-0040) pueda reconciliar offsets con spans NER
sin reparsing.

Refs #0037

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-30 16:41:30 +02:00

45 lines
1.1 KiB
Python

"""Extrae wallets BTC y ETH de un texto, con offsets."""
import re
_BTC_LEGACY = re.compile(
r"(?<![A-Za-z0-9])"
r"[13][1-9A-HJ-NP-Za-km-z]{25,34}"
r"(?![A-Za-z0-9])"
)
_BTC_BECH32 = re.compile(
r"(?<![A-Za-z0-9])"
r"bc1[02-9ac-hj-np-z]{6,87}"
r"(?![A-Za-z0-9])"
)
_ETH_REGEX = re.compile(
r"(?<![A-Za-z0-9])"
r"0x[a-fA-F0-9]{40}"
r"(?![A-Za-z0-9])"
)
def extract_crypto_wallets(text: str) -> list[dict]:
"""Extrae direcciones BTC (legacy + bech32) y ETH con offsets.
BTC legacy (P2PKH/P2SH) empieza por `1` o `3`. BTC bech32 (segwit)
empieza por `bc1`. ETH es `0x` seguido de 40 caracteres hex. No se
valida checksum — la regex es estructural.
"""
results = []
for regex, asset in (
(_BTC_LEGACY, "btc"),
(_BTC_BECH32, "btc"),
(_ETH_REGEX, "eth"),
):
for m in regex.finditer(text):
results.append({
"value": m.group(0),
"start": m.start(),
"end": m.end(),
"type": "crypto_wallet",
"asset": asset,
})
results.sort(key=lambda r: r["start"])
return results