Compare commits

..

2 Commits

Author SHA1 Message Date
nicoco
e45901064c replace setup.py and mypy.ini with pyproject.toml 2023-07-23 16:04:50 +02:00
nicoco
0b6496a7b1 remove unused CI conf files 2023-07-23 15:34:57 +02:00
13 changed files with 179 additions and 412 deletions

View File

@@ -1,22 +0,0 @@
# .readthedocs.yaml
# Read the Docs configuration file
# See https://docs.readthedocs.io/en/stable/config-file/v2.html for details
# Required
version: 2
# Set the version of Python and other tools you might need
build:
os: ubuntu-22.04
tools:
python: "3.11"
# Build documentation in the docs/ directory with Sphinx
sphinx:
configuration: docs/conf.py
# We recommend specifying your dependencies to enable reproducible builds:
# https://docs.readthedocs.io/en/stable/guides/reproducible-builds.html
python:
install:
- requirements: docs/requirements.txt

View File

@@ -1,7 +0,0 @@
language: python
python:
- "3.7"
- "3.8-dev"
install:
- "pip install ."
script: testall.py

50
_custom_build.py Normal file
View File

@@ -0,0 +1,50 @@
"""
Cythonize the .pyx file for sdist, and compile it for wheels.
NB: produced wheels are not valid, but the sdist should be.
"""
import os
from Cython.Build import cythonize
from subprocess import check_output, CalledProcessError, DEVNULL, call
from tempfile import TemporaryFile
from setuptools.command.build_py import build_py
def check_include(library_name, header):
command = [os.environ.get("PKG_CONFIG", "pkg-config"), "--cflags", library_name]
try:
cflags = check_output(command).decode("utf-8").split()
except FileNotFoundError:
print("pkg-config not found.")
return False
except CalledProcessError:
# pkg-config already prints the missing libraries on stderr.
return False
command = [os.environ.get("CC", "cc")] + cflags + ["-E", "-"]
with TemporaryFile("w+") as c_file:
c_file.write("#include <%s>" % header)
c_file.seek(0)
try:
return call(command, stdin=c_file, stdout=DEVNULL, stderr=DEVNULL) == 0
except FileNotFoundError:
print("%s headers not found." % library_name)
return False
class Build(build_py):
def run(self):
self.run_command("build_ext")
return super().run()
def initialize_options(self):
super().initialize_options()
has_python_headers = check_include("python3", "Python.h")
has_stringprep_headers = check_include("libidn", "stringprep.h")
if has_python_headers and has_stringprep_headers:
self.distribution.ext_modules = cythonize("slixmpp/stringprep.pyx")
else:
print("Falling back to the slow stringprep module.")

View File

@@ -1,15 +0,0 @@
[mypy]
check_untyped_defs = False
ignore_missing_imports = True
[mypy-slixmpp.types]
ignore_errors = True
[mypy-slixmpp.thirdparty.*]
ignore_errors = True
[mypy-slixmpp.plugins.*]
ignore_errors = True
[mypy-slixmpp.plugins.base]
ignore_errors = False

67
pyproject.toml Normal file
View File

@@ -0,0 +1,67 @@
[project]
name = "slixmpp"
version = "1.8.4"
description = 'Slixmpp is an elegant Python library for XMPP (aka Jabber).'
requires-python = ">=3.7"
dependencies = [
"aiodns >= 1.0",
"pyasn1",
"pyasn1_modules",
"typing_extensions; python_version < '3.8.0'",
]
classifiers = [
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Programming Language :: Python',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Topic :: Internet :: XMPP',
'Topic :: Software Development :: Libraries :: Python Modules',
]
readme = "README.rst"
license = { file = "LICENSE" }
[[project.authors]]
name = "Florent Le Coz"
email = "louiz@louiz.org"
[project.urls]
Repository = 'https://codeberg.org/poezio/slixmpp'
[project.optional-dependencies]
XEP-0363 = ['aiohttp']
XEP-0444 = ['emoji']
XEP-0454 = ['cryptography']
Safer-XML-parsing = ['defusedxml']
[build-system]
requires = ["setuptools", "cython"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
packages = ["slixmpp"]
py-modules = ["_custom_build"]
[tool.setuptools.cmdclass]
build_py = "_custom_build.Build"
[tool.mypy]
check_untyped_defs = false
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = 'slixmpp.types'
ignore_errors = true
[[tool.mypy.overrides]]
module = 'slixmpp.thirdparty.*'
ignore_errors = true
[[tool.mypy.overrides]]
module = 'slixmpp.plugins.*'
ignore_errors = true
[[tool.mypy.overrides]]
module = 'slixmpp.plugins.base'
ignore_errors = false

103
setup.py
View File

@@ -1,103 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Copyright (C) 2007-2011 Nathanael C. Fritz
# All Rights Reserved
#
# This software is licensed as described in the README.rst and LICENSE
# file, which you should have received as part of this distribution.
import runpy
import os
from pathlib import Path
from subprocess import call, DEVNULL, check_output, CalledProcessError
from tempfile import TemporaryFile
try:
from setuptools import setup
except ImportError:
from distutils.core import setup
from run_tests import TestCommand
version_mod = runpy.run_path('slixmpp/version.py')
VERSION = version_mod['__version__']
DESCRIPTION = ('Slixmpp is an elegant Python library for XMPP (aka Jabber).')
with open('README.rst', encoding='utf8') as readme:
LONG_DESCRIPTION = readme.read()
CLASSIFIERS = [
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Programming Language :: Python',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Topic :: Internet :: XMPP',
'Topic :: Software Development :: Libraries :: Python Modules',
]
packages = [str(mod.parent) for mod in Path('slixmpp').rglob('__init__.py')]
def check_include(library_name, header):
command = [os.environ.get('PKG_CONFIG', 'pkg-config'), '--cflags', library_name]
try:
cflags = check_output(command).decode('utf-8').split()
except FileNotFoundError:
print('pkg-config not found.')
return False
except CalledProcessError:
# pkg-config already prints the missing libraries on stderr.
return False
command = [os.environ.get('CC', 'cc')] + cflags + ['-E', '-']
with TemporaryFile('w+') as c_file:
c_file.write('#include <%s>' % header)
c_file.seek(0)
try:
return call(command, stdin=c_file, stdout=DEVNULL, stderr=DEVNULL) == 0
except FileNotFoundError:
print('%s headers not found.' % library_name)
return False
HAS_PYTHON_HEADERS = check_include('python3', 'Python.h')
HAS_STRINGPREP_HEADERS = check_include('libidn', 'stringprep.h')
ext_modules = None
if HAS_PYTHON_HEADERS and HAS_STRINGPREP_HEADERS:
try:
from Cython.Build import cythonize
except ImportError:
print('Cython not found, falling back to the slow stringprep module.')
else:
ext_modules = cythonize('slixmpp/stringprep.pyx')
else:
print('Falling back to the slow stringprep module.')
setup(
name="slixmpp",
version=VERSION,
description=DESCRIPTION,
long_description=LONG_DESCRIPTION,
author='Florent Le Coz',
author_email='louiz@louiz.org',
url='https://codeberg.org/poezio/slixmpp',
license='MIT',
platforms=['any'],
package_data={'slixmpp': ['py.typed']},
packages=packages,
ext_modules=ext_modules,
install_requires=[
'aiodns>=1.0',
'pyasn1',
'pyasn1_modules',
'typing_extensions; python_version < "3.8.0"',
],
extras_require={
'XEP-0363': ['aiohttp'],
'XEP-0444 compliance': ['emoji'],
'XEP-0454': ['cryptography'],
'Safer XML parsing': ['defusedxml'],
},
classifiers=CLASSIFIERS,
cmdclass={'test': TestCommand}
)

View File

@@ -1,7 +1,7 @@
from slixmpp.plugins.base import register_plugin
from . import stanza
from .privilege import XEP_0356
from .stanza import Perm, Privilege
from slixmpp.plugins.xep_0356 import stanza
from slixmpp.plugins.xep_0356.stanza import Perm, Privilege
from slixmpp.plugins.xep_0356.privilege import XEP_0356
register_plugin(XEP_0356)

View File

@@ -1,36 +0,0 @@
import dataclasses
from collections import defaultdict
from enum import Enum
class RosterAccess(str, Enum):
NONE = "none"
GET = "get"
SET = "set"
BOTH = "both"
class MessagePermission(str, Enum):
NONE = "none"
OUTGOING = "outgoing"
class IqPermission(str, Enum):
NONE = "none"
GET = "get"
SET = "set"
BOTH = "both"
class PresencePermission(str, Enum):
NONE = "none"
MANAGED_ENTITY = "managed_entity"
ROSTER = "roster"
@dataclasses.dataclass
class Permissions:
roster = RosterAccess.NONE
message = MessagePermission.NONE
iq = defaultdict(lambda: IqPermission.NONE)
presence = PresencePermission.NONE

View File

@@ -1,16 +1,14 @@
import logging
import typing
import uuid
from collections import defaultdict
from slixmpp import JID, Iq, Message
from slixmpp import Message, JID, Iq
from slixmpp.plugins.base import BasePlugin
from slixmpp.xmlstream import StanzaBase
from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.matcher import StanzaPath
from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.plugins.xep_0356 import stanza, Privilege, Perm
from . import stanza
from .permissions import IqPermission, MessagePermission, Permissions, RosterAccess
log = logging.getLogger(__name__)
@@ -31,7 +29,7 @@ class XEP_0356(BasePlugin):
dependencies = {"xep_0297"}
stanza = stanza
granted_privileges = defaultdict(Permissions)
granted_privileges = {"roster": "none", "message": "none", "presence": "none"}
def plugin_init(self):
if not self.xmpp.is_component:
@@ -51,42 +49,32 @@ class XEP_0356(BasePlugin):
def plugin_end(self):
self.xmpp.remove_handler("Privileges")
def _handle_privilege(self, msg: StanzaBase):
def _handle_privilege(self, msg: Message):
"""
Called when the XMPP server advertise the component's privileges.
Stores the privileges in this instance's granted_privileges attribute (a dict)
and raises the privileges_advertised event
"""
permissions = self.granted_privileges[msg.get_from()]
for perm in msg["privilege"]["perms"]:
access = perm["access"]
if access == "iq":
for ns in perm["namespaces"]:
permissions.iq[ns["ns"]] = ns["type"]
elif access in _VALID_ACCESSES:
setattr(permissions, access, perm["type"])
else:
log.warning("Received an invalid privileged access: %s", access)
self.granted_privileges[perm["access"]] = perm["type"]
log.debug(f"Privileges: {self.granted_privileges}")
self.xmpp.event("privileges_advertised")
def send_privileged_message(self, msg: Message):
if (
self.granted_privileges[msg.get_from().domain].message
!= MessagePermission.OUTGOING
):
raise PermissionError(
if self.granted_privileges["message"] == "outgoing":
self._make_privileged_message(msg).send()
else:
log.error(
"The server hasn't authorized us to send messages on behalf of other users"
)
else:
self._make_privileged_message(msg).send()
def _make_privileged_message(self, msg: Message):
server = msg.get_from().domain
wrapped = self.xmpp.make_message(mto=server, mfrom=self.xmpp.boundjid.bare)
wrapped["privilege"]["forwarded"].append(msg)
return wrapped
stanza = self.xmpp.make_message(
mto=self.xmpp.server_host, mfrom=self.xmpp.boundjid.bare
)
stanza["privilege"]["forwarded"].append(msg)
return stanza
def _make_get_roster(self, jid: typing.Union[JID, str], **iq_kwargs):
return self.xmpp.make_iq_get(
@@ -118,15 +106,9 @@ class XEP_0356(BasePlugin):
:param jid: user we want to fetch the roster from
"""
if isinstance(jid, str):
jid = JID(jid)
if self.granted_privileges[jid.domain].roster not in (
RosterAccess.GET,
RosterAccess.BOTH,
):
raise PermissionError(
"The server did not grant us privileges to get rosters"
)
if self.granted_privileges["roster"] not in ("get", "both"):
log.error("The server did not grant us privileges to get rosters")
raise ValueError
else:
return await self._make_get_roster(jid).send(**send_kwargs)
@@ -155,56 +137,8 @@ class XEP_0356(BasePlugin):
},
}
"""
if isinstance(jid, str):
jid = JID(jid)
if self.granted_privileges[jid.domain].roster not in (
RosterAccess.GET,
RosterAccess.BOTH,
):
raise PermissionError(
"The server did not grant us privileges to set rosters"
)
if self.granted_privileges["roster"] not in ("set", "both"):
log.error("The server did not grant us privileges to set rosters")
raise ValueError
else:
return await self._make_set_roster(jid, roster_items).send(**send_kwargs)
async def send_privileged_iq(
self, encapsulated_iq: Iq, iq_id: typing.Optional[str] = None
):
"""
Send an IQ on behalf of a user
Caution: the IQ *must* have the jabber:client namespace
"""
iq_id = iq_id or str(uuid.uuid4())
encapsulated_iq["id"] = iq_id
server = encapsulated_iq.get_to().domain
perms = self.granted_privileges.get(server)
if not perms:
raise PermissionError(f"{server} has not granted us any privilege")
itype = encapsulated_iq["type"]
for ns in encapsulated_iq.plugins.values():
type_ = perms.iq[ns.namespace]
if type_ == IqPermission.NONE:
raise PermissionError(
f"{server} has not granted any IQ privilege for namespace {ns.namespace}"
)
elif type_ == IqPermission.BOTH:
pass
elif type_ != itype:
raise PermissionError(
f"{server} has not granted IQ {itype} privilege for namespace {ns.namespace}"
)
iq = self.xmpp.make_iq(
itype=itype,
ifrom=self.xmpp.boundjid.bare,
ito=encapsulated_iq.get_from(),
id=iq_id,
)
iq["privileged_iq"].append(encapsulated_iq)
resp = await iq.send()
return resp["privilege"]["forwarded"]["iq"]
# does not include iq access that is handled differently
_VALID_ACCESSES = {"message", "roster", "presence"}

View File

@@ -1,12 +1,13 @@
from slixmpp.stanza import Message
from slixmpp.xmlstream import (
ElementBase,
register_stanza_plugin,
)
from slixmpp.plugins.xep_0297 import Forwarded
from slixmpp.stanza import Iq, Message
from slixmpp.xmlstream import ElementBase, register_stanza_plugin
NS = "urn:xmpp:privilege:2"
class Privilege(ElementBase):
namespace = NS
namespace = "urn:xmpp:privilege:2"
name = "privilege"
plugin_attrib = "privilege"
@@ -24,40 +25,26 @@ class Privilege(ElementBase):
def presence(self):
return self.permission("presence")
def add_perm(self, access, type_):
def iq(self):
return self.permission("iq")
def add_perm(self, access, type):
# This should only be needed for servers, so maybe out of scope for slixmpp
perm = Perm()
perm["type"] = type_
perm["type"] = type
perm["access"] = access
self.append(perm)
class Perm(ElementBase):
namespace = NS
namespace = "urn:xmpp:privilege:2"
name = "perm"
plugin_attrib = "perm"
plugin_multi_attrib = "perms"
interfaces = {"type", "access"}
class NameSpace(ElementBase):
namespace = NS
name = "namespace"
plugin_attrib = "namespace"
plugin_multi_attrib = "namespaces"
interfaces = {"ns", "type"}
class PrivilegedIq(ElementBase):
namespace = NS
name = "privileged_iq"
plugin_attrib = "privileged_iq"
def register():
register_stanza_plugin(Message, Privilege)
register_stanza_plugin(Iq, Privilege)
register_stanza_plugin(Privilege, Forwarded)
register_stanza_plugin(Privilege, Perm, iterable=True)
register_stanza_plugin(Perm, NameSpace, iterable=True)
register_stanza_plugin(Iq, PrivilegedIq)

View File

@@ -1,7 +1,9 @@
import unittest
from slixmpp import Message
from slixmpp.test import SlixTest
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.plugins.xep_0356 import stanza, permissions
from slixmpp.plugins.xep_0356 import stanza
class TestPermissions(SlixTest):
@@ -10,57 +12,30 @@ class TestPermissions(SlixTest):
def testAdvertisePermission(self):
xmlstring = """
<message from='capulet.lit' to='pubsub.capulet.lit'>
<message from='capulet.net' to='pubub.capulet.lit'>
<privilege xmlns='urn:xmpp:privilege:2'>
<perm access='roster' type='both'/>
<perm access='message' type='outgoing'/>
<perm access='presence' type='managed_entity'/>
<perm access='iq' type='both'/>
</privilege>
</message>
"""
msg = self.Message()
msg["from"] = "capulet.lit"
msg["to"] = "pubsub.capulet.lit"
msg["from"] = "capulet.net"
msg["to"] = "pubub.capulet.lit"
# This raises AttributeError: 'NoneType' object has no attribute 'use_origin_id'
# msg["id"] = "id"
for access, type_ in [
("roster", permissions.RosterAccess.BOTH),
("message", permissions.MessagePermission.OUTGOING),
("presence", permissions.PresencePermission.MANAGED_ENTITY),
("iq", permissions.IqPermission.BOTH),
("roster", "both"),
("message", "outgoing"),
("presence", "managed_entity"),
]:
msg["privilege"].add_perm(access, type_)
self.check(msg, xmlstring)
def testIqPermission(self):
x = stanza.Privilege()
x["access"] = "iq"
ns = stanza.NameSpace()
ns["ns"] = "some_ns"
ns["type"] = "get"
x["perm"]["access"] = "iq"
x["perm"].append(ns)
ns = stanza.NameSpace()
ns["ns"] = "some_other_ns"
ns["type"] = "both"
x["perm"].append(ns)
self.check(
x,
"""
<privilege xmlns='urn:xmpp:privilege:2'>
<perm access='iq'>
<namespace ns='some_ns' type='get' />
<namespace ns='some_other_ns' type='both' />
</perm>
</privilege>
"""
)
nss = set()
for perm in x["perms"]:
for ns in perm["namespaces"]:
nss.add((ns["ns"], ns["type"]))
assert nss == {("some_ns", "get"), ("some_other_ns", "both")}
# Should this one work? → # AttributeError: 'Message' object has no attribute 'permission'
# self.assertEqual(msg.permission["roster"], "both")
suite = unittest.TestLoader().loadTestsFromTestCase(TestPermissions)

View File

@@ -1,7 +1,7 @@
import unittest
from slixmpp import Message, JID, Iq
from slixmpp.plugins.xep_0356 import permissions
from slixmpp import ComponentXMPP, Iq, Message
from slixmpp.roster import RosterItem
from slixmpp.test import SlixTest
@@ -9,9 +9,9 @@ class TestPermissions(SlixTest):
def setUp(self):
self.stream_start(
mode="component",
plugins=["xep_0356", "xep_0045"],
plugins=["xep_0356"],
jid="pubsub.capulet.lit",
server="capulet.lit",
server="capulet.net",
)
def testPluginEnd(self):
@@ -23,44 +23,26 @@ class TestPermissions(SlixTest):
self.assertFalse(exc)
def testGrantedPrivileges(self):
# https://xmpp.org/extensions/xep-0356.html#example-4
results = {"event": False}
x = self.xmpp["xep_0356"]
self.xmpp.add_event_handler(
"privileges_advertised", lambda msg: results.__setitem__("event", True)
)
self.recv(
"""
<message from='capulet.lit' to='pubsub.capulet.lit' id='54321'>
<message from='capulet.net' to='pubub.capulet.lit' id='54321'>
<privilege xmlns='urn:xmpp:privilege:2'>
<perm access='roster' type='both'/>
<perm access='message' type='outgoing'/>
<perm access='iq'>
<namespace ns='some_ns' type='get' />
<namespace ns='some_other_ns' type='both' />
</perm>
</privilege>
</message>
"""
)
server = JID("capulet.lit")
self.assertEqual(self.xmpp["xep_0356"].granted_privileges["roster"], "both")
self.assertEqual(
x.granted_privileges[server].roster, permissions.RosterAccess.BOTH
)
self.assertEqual(
x.granted_privileges[server].message, permissions.MessagePermission.OUTGOING
)
self.assertEqual(
x.granted_privileges[server].presence, permissions.PresencePermission.NONE
)
self.assertEqual(
x.granted_privileges[server].iq["nope"], permissions.IqPermission.NONE
)
self.assertEqual(
x.granted_privileges[server].iq["some_ns"], permissions.IqPermission.GET
)
self.assertEqual(
x.granted_privileges[server].iq["some_other_ns"], permissions.IqPermission.BOTH
self.xmpp["xep_0356"].granted_privileges["message"], "outgoing"
)
self.assertEqual(self.xmpp["xep_0356"].granted_privileges["presence"], "none")
self.assertTrue(results["event"])
def testGetRosterIq(self):
@@ -112,7 +94,7 @@ class TestPermissions(SlixTest):
def testMakeOutgoingMessage(self):
xmlstring = """
<message xmlns="jabber:component:accept" from='pubsub.capulet.lit' to='capulet.lit'>
<message xmlns="jabber:component:accept" from='pubsub.capulet.lit' to='capulet.net'>
<privilege xmlns='urn:xmpp:privilege:2'>
<forwarded xmlns='urn:xmpp:forward:0'>
<message from="juliet@capulet.lit" to="romeo@montague.lit" xmlns="jabber:client">
@@ -126,49 +108,9 @@ class TestPermissions(SlixTest):
msg["from"] = "juliet@capulet.lit"
msg["to"] = "romeo@montague.lit"
msg["body"] = "I do not hate you"
priv_msg = self.xmpp["xep_0356"]._make_privileged_message(msg)
self.check(priv_msg, xmlstring, use_values=False)
def testDetectServer(self):
msg = Message()
msg["from"] = "juliet@something"
msg["to"] = "romeo@montague.lit"
msg["body"] = "I do not hate you"
priv_msg = self.xmpp["xep_0356"]._make_privileged_message(msg)
assert priv_msg.get_to() == "something"
assert priv_msg.get_from() == "pubsub.capulet.lit"
def testIqOnBehalf(self):
iq = Iq()
iq["mucadmin_query"]["item"]["affiliation"] = "member"
iq.set_from("juliet@xxx")
iq.set_to("somemuc@conf")
iq.set_type("get")
self.xmpp["xep_0356"].granted_privileges["conf"].iq["http://jabber.org/protocol/muc#admin"] = permissions.IqPermission.BOTH
r = self.xmpp.loop.create_task(self.xmpp["xep_0356"].send_privileged_iq(iq, iq_id="0"))
self.send(
"""
<iq from="pubsub.capulet.lit"
to="juliet@xxx"
xmlns="jabber:component:accept"
type="get" id="0">
<privileged_iq xmlns='urn:xmpp:privilege:2'>
<iq xmlns='jabber:client'
type='get'
to='somemuc@conf'
from='juliet@xxx'
id="0">
<query xmlns='http://jabber.org/protocol/muc#admin'>
<item affiliation='member'/>
</query>
</iq>
</privileged_iq>
</iq>
""",
use_values=False
)
suite = unittest.TestLoader().loadTestsFromTestCase(TestPermissions)

View File

@@ -1,5 +0,0 @@
[tox]
envlist = py34
[testenv]
deps = nose
commands = nosetests --where=tests --exclude=live -i slixtest.py