import argparse
import asyncio
import inspect
import json
import logging
import os
import re
import ssl
import sys
import certifi
import requests
import websockets
# Automatically configure Python & Requests to use certifi's CA bundle
os.environ["SSL_CERT_FILE"] = certifi.where()
os.environ["REQUESTS_CA_BUNDLE"] = certifi.where()
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")
THREE_DIGIT_REGEX = re.compile(r"^\d{3}$")
PARTICIPANT_PATH_REGEX = re.compile(r"^/?callcontrol/([^/]+)/participants/([^/]+)$", re.IGNORECASE)
def get_ssl_context():
"""Builds an SSL context referencing certifi's trusted CA store."""
return ssl.create_default_context(cafile=certifi.where())
def fetch_access_token(token_url, client_id, client_key):
"""Exchanges Client ID & Secret for an OAuth2 Bearer token."""
payload = {
"client_id": client_id,
"client_secret": client_key,
"grant_type": "client_credentials",
}
headers = {"Content-Type": "application/x-www-form-urlencoded"}
response = requests.post(token_url, data=payload, headers=headers, timeout=10)
if response.status_code != 200:
raise RuntimeError(
f"Authentication failed. HTTP {response.status_code}: {response.text}"
)
return response.json()["access_token"]
def get_participant_snapshot(rest_base_url, token, dn, participant_id):
"""Queries 3CX REST API for detailed participant call state."""
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/json",
}
url = f"{rest_base_url}/{dn}/participants/{participant_id}"
try:
res = requests.get(url, headers=headers, timeout=3)
if res.status_code == 200:
return res.json()
logging.warning(f"Could not fetch participant {participant_id}: HTTP {res.status_code}")
except Exception as e:
logging.error(f"Error querying participant details: {e}")
return None
def drop_call(rest_base_url, token, dn, participant_id):
"""Sends HTTP POST to drop the participant session."""
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
drop_url = f"{rest_base_url}/{dn}/participants/{participant_id}/drop"
try:
logging.warning(f"==> DROPPING participant {participant_id} on extension {dn}...")
response = requests.post(drop_url, headers=headers, json={}, timeout=3)
if response.status_code in (200, 204):
logging.info(f"==> Call dropped successfully (HTTP {response.status_code}).")
else:
logging.error(f"==> Drop request failed: HTTP {response.status_code} - {response.text}")
except Exception as e:
logging.error(f"==> Network error sending drop command: {e}")
async def monitor_calls(fqdn, client_id, client_key, source_extension):
"""Listens on the 3CX WebSocket and terminates unauthorized calls."""
clean_fqdn = fqdn.replace("https://", "").replace("http://", "").strip("/")
token_url = f"https://{clean_fqdn}/connect/token"
ws_url = f"wss://{clean_fqdn}/callcontrol/ws"
rest_base_url = f"https://{clean_fqdn}/callcontrol"
ssl_context = get_ssl_context()
sig = inspect.signature(websockets.connect)
header_kwarg = "additional_headers" if "additional_headers" in sig.parameters else "extra_headers"
handled_participants = set()
while True:
try:
token = fetch_access_token(token_url, client_id, client_key)
headers = {"Authorization": f"Bearer {token}"}
connect_kwargs = {header_kwarg: headers, "ssl": ssl_context}
logging.info(f"Connecting to 3CX WebSocket at {ws_url}...")
async with websockets.connect(ws_url, **connect_kwargs) as ws:
logging.info(f"Connected. Monitoring calls originating from extension {source_extension}...")
async for message in ws:
try:
raw_data = json.loads(message)
except json.JSONDecodeError:
continue
# Unwrap nested event envelope: {"sequence": X, "event": {...}}
event = raw_data.get("event") if isinstance(raw_data.get("event"), dict) else raw_data
entity = event.get("entity", "")
event_type = event.get("event_type")
match = PARTICIPANT_PATH_REGEX.match(entity)
if not match:
continue
path_dn, participant_id = match.group(1), match.group(2)
# Event type 1 = participant disconnected / removed
if event_type == 1:
handled_participants.discard(participant_id)
continue
if path_dn != source_extension:
continue
if participant_id in handled_participants:
continue
handled_participants.add(participant_id)
# Fetch active call details to identify the dialed party
snapshot = get_participant_snapshot(rest_base_url, token, path_dn, participant_id)
if not snapshot:
continue
party_dn = (
snapshot.get("party_dn")
or snapshot.get("party_caller_id")
or snapshot.get("destination")
)
logging.info(f"Extension {path_dn} dialed: {party_dn}")
# Drop if destination matches any 3-digit extension
if party_dn and THREE_DIGIT_REGEX.match(str(party_dn)):
logging.warning(f"UNAUTHORIZED CALL: {path_dn} dialed internal extension {party_dn}!")
drop_call(rest_base_url, token, path_dn, participant_id)
except websockets.exceptions.ConnectionClosed as e:
logging.error(f"WebSocket closed ({e}). Reconnecting in 5 seconds...")
await asyncio.sleep(5)
except Exception as e:
logging.error(f"Error: {e}. Retrying in 5 seconds...")
await asyncio.sleep(5)
def parse_args():
parser = argparse.ArgumentParser(description="3CX Call Interceptor & Drop Service")
parser.add_argument("--fqdn", required=True, help="3CX PBX Fully Qualified Domain Name (e.g. pbx.example.com)")
parser.add_argument("--clientid", required=True, help="3CX API Client ID")
parser.add_argument("--clientkey", required=True, help="3CX API Client Secret/Key")
parser.add_argument("--extension", required=True, help="Extension number to monitor and restrict (e.g. 112)")
return parser.parse_args()
if __name__ == "__main__":
args = parse_args()
try:
asyncio.run(
monitor_calls(
fqdn=args.fqdn,
client_id=args.clientid,
client_key=args.clientkey,
source_extension=args.extension,
)
)
except KeyboardInterrupt:
logging.info("Service stopped manually.")
except Exception as err:
logging.critical(f"Fatal error: {err}")
sys.exit(1)
