103 lines
2.5 KiB
Python
103 lines
2.5 KiB
Python
|
import ipaddress
|
||
|
|
import re
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
|
||
|
|
def digits_only(value: Optional[str]) -> str:
|
||
|
|
if not value:
|
||
|
|
return ""
|
||
|
|
return re.sub(r"\D+", "", value)
|
||
|
|
|
||
|
|
|
||
|
|
def normalize_e164(number: Optional[str]) -> Optional[str]:
|
||
|
|
if not number:
|
||
|
|
return None
|
||
|
|
|
||
|
|
raw = number.strip().replace(" ", "").replace("-", "")
|
||
|
|
if not raw:
|
||
|
|
return None
|
||
|
|
|
||
|
|
if raw.startswith("+"):
|
||
|
|
n = "+" + digits_only(raw)
|
||
|
|
return n if len(n) >= 9 else None
|
||
|
|
|
||
|
|
if raw.startswith("0045"):
|
||
|
|
rest = digits_only(raw[4:])
|
||
|
|
return "+45" + rest if len(rest) == 8 else ("+" + digits_only(raw) if len(digits_only(raw)) >= 9 else None)
|
||
|
|
|
||
|
|
d = digits_only(raw)
|
||
|
|
if len(d) == 8:
|
||
|
|
return "+45" + d
|
||
|
|
|
||
|
|
if d.startswith("45") and len(d) == 10:
|
||
|
|
return "+" + d
|
||
|
|
|
||
|
|
# Generic international (best-effort)
|
||
|
|
if len(d) >= 9:
|
||
|
|
return "+" + d
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def phone_suffix_8(number: Optional[str]) -> Optional[str]:
|
||
|
|
d = digits_only(number)
|
||
|
|
if len(d) < 8:
|
||
|
|
return None
|
||
|
|
return d[-8:]
|
||
|
|
|
||
|
|
|
||
|
|
def is_outbound_call(caller: Optional[str], local_extension: Optional[str]) -> bool:
|
||
|
|
caller_d = digits_only(caller)
|
||
|
|
local_d = digits_only(local_extension)
|
||
|
|
if not caller_d or not local_d:
|
||
|
|
return False
|
||
|
|
return caller_d.endswith(local_d)
|
||
|
|
|
||
|
|
|
||
|
|
def extract_extension(local_value: Optional[str]) -> Optional[str]:
|
||
|
|
if not local_value:
|
||
|
|
return None
|
||
|
|
|
||
|
|
raw = local_value.strip()
|
||
|
|
if not raw:
|
||
|
|
return None
|
||
|
|
|
||
|
|
if raw.isdigit():
|
||
|
|
return raw
|
||
|
|
|
||
|
|
# Common SIP format: sip:204_99773@pbx.sipserver.dk -> 204
|
||
|
|
sip_match = re.search(r"sip:([0-9]+)", raw, flags=re.IGNORECASE)
|
||
|
|
if sip_match:
|
||
|
|
return sip_match.group(1)
|
||
|
|
|
||
|
|
# Fallback: first digit run (at least 2 chars)
|
||
|
|
generic = re.search(r"([0-9]{2,})", raw)
|
||
|
|
if generic:
|
||
|
|
return generic.group(1)
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def ip_in_whitelist(client_ip: str, whitelist_csv: str) -> bool:
|
||
|
|
if not client_ip or not whitelist_csv:
|
||
|
|
return False
|
||
|
|
|
||
|
|
try:
|
||
|
|
ip_obj = ipaddress.ip_address(client_ip)
|
||
|
|
except ValueError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
for entry in [e.strip() for e in whitelist_csv.split(",") if e.strip()]:
|
||
|
|
try:
|
||
|
|
if "/" in entry:
|
||
|
|
net = ipaddress.ip_network(entry, strict=False)
|
||
|
|
if ip_obj in net:
|
||
|
|
return True
|
||
|
|
else:
|
||
|
|
if ip_obj == ipaddress.ip_address(entry):
|
||
|
|
return True
|
||
|
|
except ValueError:
|
||
|
|
continue
|
||
|
|
|
||
|
|
return False
|