mirror of
https://gitverse.ru/kpa39l/icq.git
synced 2026-09-29 01:50:10 +00:00
estorozhenko
527584281d
- index.html: omemo_default=true (XEP-0384, встроен в Converse v14) - prosody.cfg.lua: admin_web + bosh (глобальные), MUC: vcard_muc (XEP-0153) + muc_moderation (XEP-0425) - modules/: mod_admin_web (+www_files), mod_http_upload_external, mod_muc_moderation, mod_vcard_muc — из prosody-modules (apt) - scripts/omemo_check.py — проверка OMEMO-бандлов - backup.sh — ежедневный бэкап проекта (data/config/certs/modules/webchat/доки) на /mnt/yandex-disk, ротация 7/30 дней - STATUS/WALKTHROUGH: OMEMO, admin_web, HTTP Upload итоги
122 lines
5.1 KiB
Python
122 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Проверка OMEMO-связки на сервере nixg.ru (XEP-0384, PEP).
|
|
|
|
Что делает:
|
|
- логинится admin@nixg.ru по WebSocket (wss://xmpp.nixg.ru/xmpp-websocket);
|
|
- читает OMEMO devicelist (eu.siacs.conversations.axolotl.devicelist);
|
|
- по желанию публикует новое устройство (--publish <id>);
|
|
- по желанию чистит список (--clean, оставляет только реальные устройства).
|
|
|
|
Использование:
|
|
PASSWORD=... /opt/icq/.venv/bin/python /opt/icq/scripts/omemo_check.py [--publish 7777] [--clean]
|
|
JID=admin@nixg.ru (по умолчанию) — можно переопределить env JID.
|
|
|
|
Зависимости: venv /opt/icq/.venv (slixmpp 1.17).
|
|
Служебные находки (проверено 2026-08-29):
|
|
- slixmpp 1.17: connect((url,)) сам управляет циклом, process() НЕТ;
|
|
- плагины регистрировать явно: register_plugin('xep_0060');
|
|
- publish payload = XML-элемент (slixmpp.xmlstream.ET), не строка;
|
|
- get_items через WS иногда возвращает пусто — смотреть запись на диске
|
|
data/nixg%2eru/pep_eu%2esiacs%2econversations%2eaxolotl%2edevicelist/admin.list
|
|
"""
|
|
import asyncio, os, ssl, sys
|
|
sys.path.insert(0, '/opt/icq/.venv/lib/python3.12/site-packages')
|
|
import slixmpp
|
|
from slixmpp.exceptions import IqError, IqTimeout
|
|
from slixmpp.xmlstream import ET
|
|
|
|
JID = os.environ.get('JID', 'admin@nixg.ru')
|
|
PASSWORD = os.environ.get('PASSWORD', '')
|
|
if not PASSWORD:
|
|
print('ОШИБКА: задайте PASSWORD (env)'); sys.exit(2)
|
|
|
|
WS_URL = 'wss://xmpp.nixg.ru/xmpp-websocket'
|
|
NS_DEVICELIST = 'eu.siacs.conversations.axolotl.devicelist'
|
|
NS_OMEMO = 'eu.siacs.conversations.axolotl'
|
|
|
|
ctx = ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE
|
|
|
|
|
|
class OmemoCheck(slixmpp.ClientXMPP):
|
|
def __init__(self, do_publish=None, do_clean=False):
|
|
super().__init__(JID, PASSWORD)
|
|
self.register_plugin('xep_0060')
|
|
self.do_publish = do_publish
|
|
self.do_clean = do_clean
|
|
self.add_event_handler('session_start', self.go)
|
|
|
|
async def read_devicelist(self):
|
|
try:
|
|
iq = await self['xep_0060'].get_items(jid=JID, node=NS_DEVICELIST, max_items='')
|
|
its = iq['pubsub']['items']['item']
|
|
ids = []
|
|
for it in (its or []):
|
|
for dev in it.xml.iter('{%s}device' % NS_OMEMO):
|
|
ids.append(dev.get('id'))
|
|
print('[read] devicelist:', ids or '(пусто — смотри на диске, см. доки)')
|
|
return ids
|
|
except IqError as ex:
|
|
print('[read] IqError:', ex.iq['error']['condition']); return None
|
|
except Exception as ex:
|
|
print('[read]', type(ex).__name__, ex); return None
|
|
|
|
async def go(self, event):
|
|
self.send_presence(); await self.get_roster()
|
|
print('[session] OK')
|
|
ids = await self.read_devicelist()
|
|
|
|
if self.do_publish:
|
|
lst = ET.Element('{%s}list' % NS_OMEMO)
|
|
for d in (ids or []):
|
|
ET.SubElement(lst, '{%s}device' % NS_OMEMO, {'id': d})
|
|
ET.SubElement(lst, '{%s}device' % NS_OMEMO, {'id': self.do_publish})
|
|
try:
|
|
await self['xep_0060'].publish(jid=JID, node=NS_DEVICELIST, id='current', payload=lst)
|
|
print('[publish]', self.do_publish, 'ACCEPTED')
|
|
except IqError as ex:
|
|
print('[publish] IqError:', ex.iq['error']['condition'])
|
|
except IqTimeout:
|
|
print('[publish] timeout')
|
|
except Exception as ex:
|
|
print('[publish]', type(ex).__name__, ex)
|
|
|
|
if self.do_clean:
|
|
keep = [d for d in (ids or []) if d != self.do_publish] if self.do_publish else (ids or [])
|
|
# без явного списка чистить неоткуда — просто ре-публикуем текущий
|
|
lst = ET.Element('{%s}list' % NS_OMEMO)
|
|
for d in keep:
|
|
ET.SubElement(lst, '{%s}device' % NS_OMEMO, {'id': d})
|
|
try:
|
|
await self['xep_0060'].publish(jid=JID, node=NS_DEVICELIST, id='current', payload=lst)
|
|
print('[clean] devicelist ->', keep, 'ACCEPTED')
|
|
except Exception as ex:
|
|
print('[clean]', type(ex).__name__, ex)
|
|
|
|
await asyncio.sleep(1)
|
|
self.disconnect()
|
|
|
|
|
|
async def main():
|
|
do_publish = None
|
|
do_clean = False
|
|
args = sys.argv[1:]
|
|
if '--publish' in args:
|
|
do_publish = args[args.index('--publish') + 1]
|
|
if '--clean' in args:
|
|
do_clean = True
|
|
|
|
bot = OmemoCheck(do_publish=do_publish, do_clean=do_clean)
|
|
bot.ssl_context = ctx
|
|
try:
|
|
fut = bot.connect((WS_URL,))
|
|
await asyncio.wait_for(fut, timeout=25)
|
|
await asyncio.sleep(8)
|
|
try: bot.disconnect()
|
|
except Exception: pass
|
|
except Exception as ex:
|
|
print('connect error:', ex)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|