XEP-0385: minimal support

- includes bits of other required XEPs
- only implements 'outgoing' SIMS

slixmpp/plugins/xep_0234/__init__.py

@@ -0,0 +1,6 @@
+from slixmpp.plugins.base import register_plugin
+
+from . import stanza
+from .jingle_file_transfer import XEP_0234
+
+register_plugin(XEP_0234)

slixmpp/plugins/xep_0234/jingle_file_transfer.py

@@ -0,0 +1,21 @@
+import logging
+
+from slixmpp.plugins import BasePlugin
+
+from . import stanza
+
+log = logging.getLogger(__name__)
+
+
+class XEP_0234(BasePlugin):
+
+    """
+    XEP-0234: Jingle File Transfer
+
+    Minimum needed for xep 0385 (Stateless inline media sharing)
+    """
+
+    name = "xep_0234"
+    description = "XEP-0234: Jingle File Transfer"
+    dependencies = {"xep_0082", "xep_0300"}
+    stanza = stanza

slixmpp/plugins/xep_0234/stanza.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+
+from slixmpp.plugins.xep_0082 import format_datetime, parse
+from slixmpp.xmlstream import ElementBase
+
+NS = "urn:xmpp:jingle:apps:file-transfer:5"
+
+
+class File(ElementBase):
+    name = "file"
+    namespace = NS
+    plugin_attrib = "file"
+    interfaces = sub_interfaces = {"media-type", "name", "date", "size", "hash", "desc"}
+
+    def set_size(self, size: int):
+        self._set_sub_text("size", str(size))
+
+    def get_size(self):
+        return _int_or_none(self._get_sub_text("size"))
+
+    def get_date(self):
+        try:
+            return parse(self._get_sub_text("date"))
+        except ValueError:
+            return
+
+    def set_date(self, stamp: datetime):
+        try:
+            self._set_sub_text("date", format_datetime(stamp))
+        except ValueError:
+            pass
+
+
+def _int_or_none(v):
+    try:
+        return int(v)
+    except ValueError:
+        return None

slixmpp/plugins/xep_0372/__init__.py

@@ -0,0 +1,6 @@
+from slixmpp.plugins.base import register_plugin
+
+from . import stanza
+from .references import XEP_0372
+
+register_plugin(XEP_0372)

slixmpp/plugins/xep_0372/references.py

@@ -0,0 +1,23 @@
+import logging
+
+from slixmpp import Message, register_stanza_plugin
+from slixmpp.plugins import BasePlugin
+
+from . import stanza
+
+log = logging.getLogger(__name__)
+
+
+class XEP_0372(BasePlugin):
+    """
+    XEP-0372: References
+
+    Minimum needed for xep 0385 (Stateless inline media sharing)
+    """
+
+    name = "xep_0372"
+    description = "XEP-0372: References"
+    stanza = stanza
+
+    def plugin_init(self):
+        register_stanza_plugin(Message, stanza.Reference)

slixmpp/plugins/xep_0372/stanza.py

@@ -0,0 +1,9 @@
+from slixmpp.xmlstream import ElementBase
+
+NAMESPACE = "urn:xmpp:reference:0"
+
+
+class Reference(ElementBase):
+    name = plugin_attrib = "reference"
+    namespace = NAMESPACE
+    interfaces = {"type", "uri", "id", "begin", "end"}

slixmpp/plugins/xep_0385/__init__.py

@@ -0,0 +1,11 @@
+
+# Slixmpp: The Slick XMPP Library
+# Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout
+# This file is part of Slixmpp.
+# See the file LICENSE for copying permission
+from slixmpp.plugins.base import register_plugin
+
+from . import stanza
+from .sims import XEP_0385
+
+register_plugin(XEP_0385)

slixmpp/plugins/xep_0385/sims.py

@@ -0,0 +1,66 @@
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Iterable, Optional
+
+from slixmpp.plugins import BasePlugin
+from slixmpp.stanza import Message
+from slixmpp.xmlstream import register_stanza_plugin
+
+from . import stanza
+
+log = logging.getLogger(__name__)
+
+
+class XEP_0385(BasePlugin):
+
+    """
+    XEP-0385: Stateless Inline Media Sharing (SIMS)
+
+    Only support outgoing SIMS, incoming is not handled at all.
+    """
+
+    name = "xep_0385"
+    description = "XEP-0385: Stateless Inline Media Sharing (SIMS)"
+    dependencies = {"xep_0234", "xep_0300", "xep_0372"}
+    stanza = stanza
+
+    def plugin_init(self):
+        register_stanza_plugin(self.xmpp["xep_0372"].stanza.Reference, stanza.Sims)
+        register_stanza_plugin(Message, stanza.Sims)
+
+        register_stanza_plugin(stanza.Sims, stanza.Sources)
+        register_stanza_plugin(stanza.Sims, self.xmpp["xep_0234"].stanza.File)
+        register_stanza_plugin(stanza.Sources, self.xmpp["xep_0372"].stanza.Reference)
+
+    def get_sims(
+        self,
+        path: Path,
+        uris: Iterable[str],
+        media_type: Optional[str],
+        desc: Optional[str],
+    ):
+        sims = stanza.Sims()
+        for uri in uris:
+            ref = self.xmpp["xep_0372"].stanza.Reference()
+            ref["uri"] = uri
+            ref["type"] = "data"
+            sims["sources"].append(ref)
+        if media_type:
+            sims["file"]["media-type"] = media_type
+        if desc:
+            sims["file"]["desc"] = desc
+        sims["file"]["name"] = path.name
+
+        stat = path.stat()
+        sims["file"]["size"] = stat.st_size
+        sims["file"]["date"] = datetime.fromtimestamp(stat.st_mtime)
+
+        h = self.xmpp.plugin["xep_0300"].compute_hash(path)
+        h["value"] = h["value"].decode()
+        sims["file"].append(h)
+
+        ref = self.xmpp["xep_0372"].stanza.Reference()
+        ref.append(sims)
+        ref["type"] = "data"
+        return ref

slixmpp/plugins/xep_0385/stanza.py

@@ -0,0 +1,14 @@
+from slixmpp.xmlstream import ElementBase
+
+NAMESPACE = "urn:xmpp:sims:1"
+
+
+class Sims(ElementBase):
+    name = "media-sharing"
+    plugin_attrib = "sims"
+    namespace = NAMESPACE
+
+
+class Sources(ElementBase):
+    name = plugin_attrib = "sources"
+    namespace = NAMESPACE

tests/test_stream_xep_0385.py

@@ -0,0 +1,60 @@
+import unittest
+from base64 import b64encode
+from datetime import datetime
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from hashlib import sha256
+
+from slixmpp.plugins.xep_0082 import format_datetime
+from slixmpp.test import SlixTest
+
+class TestSIMS(SlixTest):
+    def setUp(self):
+        self.stream_start(
+            mode="component", jid="whatevs.shakespeare.lit", plugins={"xep_0385"}
+        )
+
+    def tearDown(self):
+        self.stream_close()
+
+    def test_set_file(self):
+        with NamedTemporaryFile("wb+") as f:
+            n = 10
+            size = 0
+            for i in range(n):
+                size += len(bytes(i))
+                f.write(bytes(i))
+
+            f.seek(0)
+            h = b64encode(sha256(f.read()).digest()).decode()
+            sims = self.xmpp["xep_0385"].get_sims(
+                Path(f.name),
+                ["https://xxx.com"],
+                media_type="MEDIA",
+                desc="DESCRIPTION",
+            )
+
+            self.check(
+                sims,
+                f"""
+                <reference xmlns='urn:xmpp:reference:0' type='data'>
+                <media-sharing xmlns='urn:xmpp:sims:1'>
+                  <file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
+                    <media-type>MEDIA</media-type>
+                    <name>{Path(f.name).name}</name>
+                    <size>{size}</size>
+                    <hash xmlns='urn:xmpp:hashes:2' algo='sha-256'>{h}</hash>
+                    <desc>DESCRIPTION</desc>
+                    <date>{format_datetime(datetime.fromtimestamp(Path(f.name).stat().st_mtime))}</date>
+                  </file>
+                  <sources>
+                    <reference xmlns='urn:xmpp:reference:0' type='data' uri='https://xxx.com' />
+                  </sources>
+                </media-sharing>
+                </reference>
+                """,
+                use_values=False,
+            )
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestSIMS)