mirror of
https://gitverse.ru/kpa39l/icq.git
synced 2026-09-29 09:55:11 +00:00
83 lines
3.8 KiB
Python
83 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Функциональный тест mod_privilege (XEP-0356) на Prosody 13.0.
|
||
|
||
Подключается как внешний компонент telegram.test.nixg.ru (XEP-0114),
|
||
затем от имени компонента делает привилегированный запрос roster
|
||
пользователя testuser1@test.nixg.ru.
|
||
Если mod_privilege работает — компонент получит roster.
|
||
Если нет — сервер ответит forbidden/not-allowed.
|
||
"""
|
||
import asyncio, logging, ssl
|
||
import slixmpp
|
||
from slixmpp import ComponentXMPP
|
||
from slixmpp.xmlstream import ET
|
||
from slixmpp.xmlstream.handler import Callback
|
||
from slixmpp.xmlstream.matcher import MatchXPath
|
||
|
||
logging.basicConfig(level=logging.WARNING, format='%(levelname)s %(name)s: %(message)s')
|
||
logging.getLogger('slixmpp.xmlstream.xmlstream').setLevel(logging.DEBUG)
|
||
|
||
async def main():
|
||
comp = ComponentXMPP("telegram.test.nixg.ru", "test-secret-telegram-123",
|
||
"test.nixg.ru")
|
||
comp.ssl_context = ssl.create_default_context()
|
||
comp.ssl_context.check_hostname = False
|
||
comp.ssl_context.verify_mode = ssl.CERT_NONE
|
||
|
||
result = {"done": False}
|
||
|
||
def on_iq(ev):
|
||
# Callback на {jabber:component:accept}iq присылает строку XML, парсим
|
||
if isinstance(ev, str):
|
||
ev = ET.fromstring(ev)
|
||
print(f">>> on_iq fired: type={ev.get('type')}, id={ev.get('id')}")
|
||
if ev.get('type') in ('result', 'error') and ev.get('id') == 'priv-roster-1':
|
||
result['done'] = True
|
||
if ev.get('type') == 'result':
|
||
print(">>> PRIVILEGE OK: mod_privilege ответил result (roster testuser1)")
|
||
# достаём все <item> из <query>
|
||
for q in ev.findall('{jabber:iq:roster}query'):
|
||
for it in q.findall('{jabber:iq:roster}item'):
|
||
print(f" - {it.get('jid')} (sub={it.get('subscription')})")
|
||
if not ev.findall('{jabber:iq:roster}query/*'):
|
||
print(" (roster пуст — у testuser1 нет контактов)")
|
||
else:
|
||
conds = ev.findall('{urn:ietf:params:xml:ns:xmpp-stanzas}*')
|
||
cond = conds[0].tag.split('}')[1] if conds else '?'
|
||
print(f">>> PRIVILEGE FAILED: {cond}")
|
||
comp.disconnect()
|
||
|
||
comp.add_event_handler('iq', on_iq)
|
||
|
||
def on_session_start(ev):
|
||
print(">>> component connected as telegram.test.nixg.ru")
|
||
# Привилегированный запрос: получить roster testuser1
|
||
iq = comp.Iq()
|
||
iq['id'] = 'priv-roster-1'
|
||
iq['type'] = 'get'
|
||
iq['to'] = 'testuser1@test.nixg.ru'
|
||
iq['from'] = 'telegram.test.nixg.ru'
|
||
iq.append(ET.Element('{jabber:iq:roster}query'))
|
||
comp.send(iq)
|
||
|
||
comp.add_event_handler('session_start', on_session_start)
|
||
comp.add_event_handler('disconnected', lambda ev: print(">>> component disconnected"))
|
||
comp.add_event_handler('connection_failed', lambda ev: print(">>> connection_failed", ev))
|
||
|
||
# В slixmpp 1.17 ComponentXMPP НЕ обрабатывает входящие IQ (только handshake + presence_probe) —
|
||
# регистрируем свой Callback на IQ, чтобы поймать ответ mod_privilege
|
||
def _on_iq_cb(iq):
|
||
on_iq(iq)
|
||
|
||
comp.register_handler(
|
||
Callback('IQPriv', MatchXPath('{jabber:component:accept}iq'), _on_iq_cb))
|
||
|
||
try:
|
||
await asyncio.wait_for(comp.connect('127.0.0.1', 15347), 15)
|
||
await asyncio.wait_for(comp.wait_until('disconnected'), 30)
|
||
except asyncio.TimeoutError:
|
||
print(">>> TIMEOUT (нет ответа от mod_privilege)")
|
||
except Exception as e:
|
||
print(f">>> EXC: {type(e).__name__}: {e}")
|
||
|
||
asyncio.run(main()) |