[feat] Watchdog and notifications (#2895)
This commit is contained in:
+70
-10
@@ -1,13 +1,22 @@
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
from typing import Dict, Union
|
||||
import re
|
||||
from typing import Dict, Tuple, Union
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import secp256k1
|
||||
from bech32 import bech32_decode, bech32_encode, convertbits
|
||||
from Cryptodome import Random
|
||||
from Cryptodome.Cipher import AES
|
||||
from Cryptodome.Util.Padding import pad, unpad
|
||||
from pynostr.key import PrivateKey
|
||||
|
||||
|
||||
def generate_keypair() -> Tuple[str, str]:
|
||||
private_key = PrivateKey()
|
||||
public_key = private_key.public_key
|
||||
return private_key.hex(), public_key.hex()
|
||||
|
||||
|
||||
def encrypt_content(
|
||||
@@ -155,22 +164,30 @@ def json_dumps(data: Union[Dict, list]) -> str:
|
||||
return json.dumps(data, separators=(",", ":"), ensure_ascii=False)
|
||||
|
||||
|
||||
def normalize_public_key(pubkey: str) -> str:
|
||||
if pubkey.startswith("npub1"):
|
||||
_, decoded_data = bech32_decode(pubkey)
|
||||
assert decoded_data, "Public Key is not valid npub."
|
||||
def normalize_public_key(key: str) -> str:
|
||||
return normalize_bech32_key("npub1", key)
|
||||
|
||||
|
||||
def normalize_private_key(key: str) -> str:
|
||||
return normalize_bech32_key("nsec1", key)
|
||||
|
||||
|
||||
def normalize_bech32_key(hrp: str, key: str) -> str:
|
||||
if key.startswith(hrp):
|
||||
_, decoded_data = bech32_decode(key)
|
||||
assert decoded_data, f"Key is not valid {hrp}."
|
||||
|
||||
decoded_data_bits = convertbits(decoded_data, 5, 8, False)
|
||||
assert decoded_data_bits, "Public Key is not valid npub."
|
||||
assert decoded_data_bits, f"Key is not valid {hrp}."
|
||||
|
||||
return bytes(decoded_data_bits).hex()
|
||||
|
||||
assert len(pubkey) == 64, "Public key has wrong length."
|
||||
assert len(key) == 64, "Key has wrong length."
|
||||
try:
|
||||
int(pubkey, 16)
|
||||
int(key, 16)
|
||||
except Exception as exc:
|
||||
raise AssertionError("Public Key is not valid hex.") from exc
|
||||
return pubkey
|
||||
raise AssertionError("Key is not valid hex.") from exc
|
||||
return key
|
||||
|
||||
|
||||
def hex_to_npub(hex_pubkey: str) -> str:
|
||||
@@ -188,3 +205,46 @@ def hex_to_npub(hex_pubkey: str) -> str:
|
||||
bits = convertbits(pubkey_bytes, 8, 5, True)
|
||||
assert bits
|
||||
return bech32_encode("npub", bits)
|
||||
|
||||
|
||||
def normalize_identifier(identifier: str):
|
||||
identifier = identifier.lower().split("@")[0]
|
||||
validate_identifier(identifier)
|
||||
return identifier
|
||||
|
||||
|
||||
def validate_pub_key(pubkey: str) -> str:
|
||||
if pubkey.startswith("npub"):
|
||||
_, data = bech32_decode(pubkey)
|
||||
if data:
|
||||
decoded_data = convertbits(data, 5, 8, False)
|
||||
if decoded_data:
|
||||
pubkey = bytes(decoded_data).hex()
|
||||
try:
|
||||
_hex = bytes.fromhex(pubkey)
|
||||
except Exception as exc:
|
||||
raise ValueError("Pubkey must be in npub or hex format.") from exc
|
||||
|
||||
if len(_hex) != 32:
|
||||
raise ValueError("Pubkey length incorrect.")
|
||||
|
||||
return pubkey
|
||||
|
||||
|
||||
def validate_identifier(local_part: str):
|
||||
regex = re.compile(r"^[a-z0-9_.]+$")
|
||||
if not re.fullmatch(regex, local_part.lower()):
|
||||
raise ValueError(
|
||||
f"Identifier '{local_part}' not allowed! "
|
||||
"Only a-z, 0-9 and .-_ are allowed characters, case insensitive."
|
||||
)
|
||||
|
||||
|
||||
def is_ws_url(url):
|
||||
try:
|
||||
result = urlparse(url)
|
||||
if not all([result.scheme, result.netloc]):
|
||||
return False
|
||||
return result.scheme in ["ws", "wss"]
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
Reference in New Issue
Block a user