3cx-telefonie-middleware/tools/3cx_abandoned.py

#!/usr/bin/env python3
import base64
import re
import requests
from dataclasses import dataclass

URL = "https://schmid-gartenpflanzen.on3cx.de/MyPhone/MPWebService.asmx"
SESSION = "b7a74b60-7987-d1ae-9189-49dd1c48803a"

@dataclass
class AbandonedCall:
    call_id: int
    number: str | None
    queue: str | None
    raw: bytes

def read_varint(data, pos):
    value = 0
    shift = 0
    while pos < len(data):
        b = data[pos]
        pos += 1
        value |= (b & 0x7f) << shift
        if not b & 0x80:
            return value, pos
        shift += 7
    raise ValueError("unterminated varint")

def fields(data):
    pos = 0
    while pos < len(data):
        key, pos = read_varint(data, pos)
        field_no = key >> 3
        wire = key & 7

        if wire == 0:
            value, pos = read_varint(data, pos)
        elif wire == 2:
            length, pos = read_varint(data, pos)
            value = data[pos:pos + length]
            pos += length
        elif wire == 1:
            value = data[pos:pos + 8]
            pos += 8
        elif wire == 5:
            value = data[pos:pos + 4]
            pos += 4
        else:
            raise ValueError(f"unsupported wire type {wire}")

        yield field_no, wire, value

def strings(data):
    out = []
    for field_no, wire, value in fields(data):
        if wire == 2:
            try:
                text = value.decode("utf-8")
                if text.isprintable() and len(text) >= 2:
                    out.append((field_no, text))
            except UnicodeDecodeError:
                pass
    return out

def find_call_messages(data):
    """
    Command 145 liefert eine Liste verschachtelter Call-Messages.
    Wir suchen rekursiv nach Messages, die eine E.164-Nummer enthalten.
    """
    results = []

    def walk(blob):
        try:
            fs = list(fields(blob))
        except Exception:
            return

        text = strings(blob)
        numbers = [
            value for _, value in text
            if re.fullmatch(r"\+\d{6,20}", value)
        ]

        if numbers:
            # bekannte Struktur: irgendwo in derselben Message
            # liegt die numerische Call-ID als varint.
            varints = [
                value for _, wire, value in fs
                if wire == 0 and isinstance(value, int)
            ]

            # IDs wie 11626/11627 sind hier besonders interessant.
            ids = [v for v in varints if 1000 <= v <= 10000000]

            if ids:
                results.append((ids[0], numbers[0], blob))

        for _, wire, value in fs:
            if wire == 2 and isinstance(value, bytes) and len(value) > 2:
                walk(value)

    walk(data)
    return results

def main():
    # Command 145, beobachteter Request:
    # 08 91 01 8a 09 06 10 50 18 00 22 00
    payload = bytes.fromhex("08 91 01 8a 09 06 10 50 18 00 22 00")

    r = requests.post(
        URL,
        headers={
            "Accept": "application/octet-stream",
            "Content-Type": "application/octet-stream",
            "MyPhoneSession": SESSION,
            "Origin": "https://schmid-gartenpflanzen.on3cx.de",
        },
        data=payload,
        verify=False,
        timeout=15,
    )
    r.raise_for_status()

    print(f"HTTP: {r.status_code}")
    print(f"Response: {len(r.content)} bytes")

    calls = find_call_messages(r.content)

    seen = set()
    for call_id, number, raw in calls:
        key = (call_id, number)
        if key in seen:
            continue
        seen.add(key)
        print(f"{call_id:>8}  {number}")

if __name__ == "__main__":
    main()