python/python-txWS
SHA256
14
0
Fork 0
forked from pool/python-txWS
Code Pull Requests Activity

Accepting request 1089605 from home:pgajdos:python

Browse Source 
- do not require six
- added patches
  fix https://github.com/MostAwesomeDude/txWS/issues/36
  + python-txWS-no-python2.patch
  05aadd036a
  + python-txWS-tobytes.patch
- added sources
  9e3a2a464b
  + tests.py

OBS-URL: https://build.opensuse.org/request/show/1089605
OBS-URL: https://build.opensuse.org/package/show/devel:languages:python/python-txWS?expand=0&rev=9
This commit is contained in:
Matej Cepl
2023-05-30 13:40:35 +00:00
committed by Git OBS Bridge
parent 757abba350
commit 10b8f8bb61
5 changed files with 546 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
python-txWS-no-python2.patch Normal file
View File

@@ -0,0 +1,191 @@
Index: txWS-0.9.1/txws.py
===================================================================
--- txWS-0.9.1.orig/txws.py
+++ txWS-0.9.1/txws.py
@@ -23,12 +23,9 @@ Blind reimplementation of WebSockets as
protocols.
"""
-from __future__ import division
__version__ = "0.7.1"
-import six
-
import array
from base64 import b64encode, b64decode
@@ -101,7 +98,7 @@ def http_headers(s):
for line in s.split("\r\n"):
try:
- key, value = [i.strip() for i in line.split(":", 1)]
+ key, value = (i.strip() for i in line.split(":", 1))
d[key] = value
except ValueError:
pass
@@ -139,7 +136,7 @@ def complete_hybi00(headers, challenge):
first = int("".join(i for i in key1 if i in digits)) // key1.count(" ")
second = int("".join(i for i in key2 if i in digits)) // key2.count(" ")
- nonce = pack(">II8s", first, second, six.b(challenge))
+ nonce = pack(">II8s", first, second, challenge)
return md5(nonce).digest()
@@ -152,7 +149,7 @@ def make_accept(key):
guid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
- accept = "%s%s" % (key, guid)
+ accept = "{}{}".format(key, guid)
hashed_bytes = sha1(accept.encode('utf-8')).digest()
return b64encode(hashed_bytes).strip().decode('utf-8')
@@ -169,10 +166,10 @@ def make_hybi00_frame(buf):
and valid text without any 0xff bytes.
"""
- if isinstance(buf, six.text_type):
+ if isinstance(buf, str):
buf = buf.encode('utf-8')
- return six.b("\x00") + buf + six.b("\xff")
+ return b"\x00" + buf + b"\xff"
def parse_hybi00_frames(buf):
"""
@@ -182,12 +179,12 @@ def parse_hybi00_frames(buf):
and will actively ignore it.
"""
- start = buf.find(six.b("\x00"))
+ start = buf.find(b"\x00")
tail = 0
frames = []
while start != -1:
- end = buf.find(six.b("\xff"), start + 1)
+ end = buf.find(b"\xff", start + 1)
if end == -1:
# Incomplete frame, try again later.
break
@@ -196,7 +193,7 @@ def parse_hybi00_frames(buf):
frame = buf[start + 1:end]
frames.append((NORMAL, frame))
tail = end + 1
- start = buf.find(six.b("\x00"), end + 1)
+ start = buf.find(b"\x00", end + 1)
# Adjust the buffer and return.
buf = buf[tail:]
@@ -231,12 +228,12 @@ def make_hybi07_frame(buf, opcode=0x1):
else:
length = chr(len(buf))
- if isinstance(buf, six.text_type):
+ if isinstance(buf, str):
buf = buf.encode('utf-8')
# Always make a normal packet.
header = chr(0x80 | opcode)
- return six.b(header + length) + buf
+ return bytes(header + length) + buf
def make_hybi07_frame_dwim(buf):
"""
@@ -244,9 +241,9 @@ def make_hybi07_frame_dwim(buf):
"""
# TODO: eliminate magic numbers.
- if isinstance(buf, six.binary_type):
+ if isinstance(buf, bytes):
return make_hybi07_frame(buf, opcode=0x2)
- elif isinstance(buf, six.text_type):
+ elif isinstance(buf, str):
return make_hybi07_frame(buf.encode("utf-8"), opcode=0x1)
else:
raise TypeError("In binary support mode, frame data must be either str or unicode")
@@ -268,9 +265,6 @@ def parse_hybi07_frames(buf):
# about, and an opcode which nobody cares about.
header = buf[start]
- if six.PY2:
- header = ord(header)
-
if header & 0x70:
# At least one of the reserved flags is set. Pork chop sandwiches!
raise WSException("Reserved flag in HyBi-07 frame (%d)" % header)
@@ -289,9 +283,6 @@ def parse_hybi07_frames(buf):
# extra length.
length = buf[start + 1]
- if six.PY2:
- length = ord(length)
-
masked = length & 0x80
length &= 0x7f
@@ -342,7 +333,7 @@ def parse_hybi07_frames(buf):
data = unpack(">H", data[:2])[0], data[2:]
else:
# No reason given; use generic data.
- data = 1000, six.b("No reason given")
+ data = 1000, b"No reason given"
frames.append((opcode, data))
start += offset + length
@@ -355,7 +346,7 @@ class WebSocketProtocol(ProtocolWrapper)
layer.
"""
- buf = six.b("")
+ buf = b""
codec = None
location = "/"
host = "example.com"
@@ -385,7 +376,7 @@ class WebSocketProtocol(ProtocolWrapper)
return ISSLTransport(self.transport, None) is not None
def writeEncoded(self, data):
- if isinstance(data, six.text_type):
+ if isinstance(data, str):
data = data.encode('utf-8')
self.transport.write(data)
@@ -418,7 +409,7 @@ class WebSocketProtocol(ProtocolWrapper)
self.writeEncodedSequence([
"Sec-WebSocket-Origin: %s\r\n" % self.origin,
- "Sec-WebSocket-Location: %s://%s%s\r\n" % (protocol, self.host,
+ "Sec-WebSocket-Location: {}://{}{}\r\n".format(protocol, self.host,
self.location),
"WebSocket-Protocol: %s\r\n" % self.codec,
"Sec-WebSocket-Protocol: %s\r\n" % self.codec,
@@ -528,7 +519,7 @@ class WebSocketProtocol(ProtocolWrapper)
elif "Sec-WebSocket-Protocol" in self.headers:
protocols = self.headers["Sec-WebSocket-Protocol"]
- if isinstance(protocols, six.string_types):
+ if isinstance(protocols, str):
protocols = [p.strip() for p in protocols.split(',')]
for protocol in protocols:
@@ -587,7 +578,7 @@ class WebSocketProtocol(ProtocolWrapper)
# These lines look like:
# GET /some/path/to/a/websocket/resource HTTP/1.1
if self.state == REQUEST:
- separator = six.b("\r\n")
+ separator = b"\r\n"
if separator in self.buf:
request, chaff, self.buf = self.buf.partition(separator)
request = request.decode('utf-8')
@@ -601,7 +592,7 @@ class WebSocketProtocol(ProtocolWrapper)
elif self.state == NEGOTIATING:
# Check to see if we've got a complete set of headers yet.
- separator = six.b("\r\n\r\n")
+ separator = b"\r\n\r\n"
if separator in self.buf:
head, chaff, self.buf = self.buf.partition(separator)
head = head.decode('utf-8')

26
python-txWS-tobytes.patch Normal file
View File

@@ -0,0 +1,26 @@
Index: txWS-0.9.1/txws.py
===================================================================
--- txWS-0.9.1.orig/txws.py
+++ txWS-0.9.1/txws.py
@@ -211,7 +211,7 @@ def mask(buf, key):
buf = array.array("B", buf)
for i in range(len(buf)):
buf[i] ^= key[i % 4]
- return buf.tostring()
+ return buf.tobytes()
def make_hybi07_frame(buf, opcode=0x1):
"""
Index: txWS-0.9.1/setup.py
===================================================================
--- txWS-0.9.1.orig/setup.py
+++ txWS-0.9.1/setup.py
@@ -5,7 +5,7 @@ from setuptools import setup
setup(
name="txWS",
py_modules=["txws"],
- setup_requires=["vcversioner", "six"],
+ setup_requires=["vcversioner"],
vcversioner={},
author="Corbin Simpson",
author_email="simpsoco@osuosl.org",

13
python-txWS.changes
View File

@@ -1,3 +1,16 @@
-------------------------------------------------------------------
Mon May 29 15:35:52 UTC 2023 - pgajdos@suse.com
- do not require six
- added patches
fix https://github.com/MostAwesomeDude/txWS/issues/36
+ python-txWS-no-python2.patch
https://github.com/MostAwesomeDude/txWS/commit/05aadd036a7d9a0959c0d915a139779706e960d7
+ python-txWS-tobytes.patch
- added sources
https://github.com/MostAwesomeDude/txWS/commit/9e3a2a464b1c908086c82b293c271e58196f83df
+ tests.py
-------------------------------------------------------------------
Tue Dec 4 12:55:29 UTC 2018 - Matej Cepl <mcepl@suse.com>

31
python-txWS.spec
View File

@@ -1,7 +1,7 @@
#
# spec file for package python-txWS
#
# Copyright (c) 2018 SUSE LINUX GmbH, Nuernberg, Germany.
# Copyright (c) 2023 SUSE LLC
#
# All modifications and additions to the file contributed by third parties
# remain the property of their copyright owners, unless otherwise agreed
@@ -16,21 +16,27 @@
#
%{?!python_module:%define python_module() python-%{**} python3-%{**}}
%bcond_without test
Name: python-txWS
Version: 0.9.1
Release: 0
Summary: Twisted WebSockets wrapper
License: X11
Group: Development/Languages/Python
URL: http://github.com/MostAwesomeDude/txWS
Source: https://files.pythonhosted.org/packages/source/t/txWS/txWS-%{version}.tar.gz
URL: https://github.com/MostAwesomeDude/txWS
Source0: https://files.pythonhosted.org/packages/source/t/txWS/txWS-%{version}.tar.gz
# https://github.com/MostAwesomeDude/txWS/commit/9e3a2a464b1c908086c82b293c271e58196f83df
Source1: https://raw.githubusercontent.com/MostAwesomeDude/txWS/master/tests.py
# https://github.com/MostAwesomeDude/txWS/issues/36
Patch0: python-txWS-no-python2.patch
# https://github.com/MostAwesomeDude/txWS/commit/05aadd036a7d9a0959c0d915a139779706e960d7
Patch1: python-txWS-tobytes.patch
BuildRequires: %{python_module Twisted}
BuildRequires: %{python_module setuptools}
BuildRequires: %{python_module six}
BuildRequires: %{python_module vcversioner}
BuildRequires: fdupes
BuildRequires: python-rpm-macros
BuildRequires: python3-pyupgrade
Requires: python-Twisted
BuildArch: noarch
%python_subpackages
@@ -39,7 +45,7 @@ txWS (Twisted WebSockets) is a library for
adding WebSockets server support to Twisted applications.
%prep
%setup -q -n txWS-%{version}
%autosetup -p1 -n txWS-%{version}
%build
%python_build
@@ -48,9 +54,18 @@ adding WebSockets server support to Twisted applications.
%python_install
%python_expand %fdupes %{buildroot}%{$python_sitelib}
%check
cp %{SOURCE1} .
# https://github.com/MostAwesomeDude/txWS/issues/36
pyupgrade tests.py || true
sed -i '/import six/d' tests.py
sed -i 's:\(challenge = \)\(.*\):\1b\2:' tests.py
%pyunittest -v
%files %{python_files}
%license LICENSE
%doc CHANGELOG.rst README.rst
%{python_sitelib}/*
%{python_sitelib}/tx{WS,ws}*
%{python_sitelib}/__pycache__
%changelog

293
tests.py Normal file
View File

@@ -0,0 +1,293 @@
# Copyright 2014 Google Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy
# of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
import six
from twisted.trial import unittest
from txws import (is_hybi00, complete_hybi00, make_hybi00_frame,
parse_hybi00_frames, http_headers, make_accept, mask, CLOSE,
NORMAL, PING, PONG, parse_hybi07_frames)
class TestHTTPHeaders(unittest.TestCase):
def test_single_header(self):
raw = "Connection: Upgrade"
headers = http_headers(raw)
self.assertTrue("Connection" in headers)
self.assertEqual(headers["Connection"], "Upgrade")
def test_single_header_newline(self):
raw = "Connection: Upgrade\r\n"
headers = http_headers(raw)
self.assertEqual(headers["Connection"], "Upgrade")
def test_multiple_headers(self):
raw = "Connection: Upgrade\r\nUpgrade: WebSocket"
headers = http_headers(raw)
self.assertEqual(headers["Connection"], "Upgrade")
self.assertEqual(headers["Upgrade"], "WebSocket")
def test_origin_colon(self):
"""
Some headers have multiple colons in them.
"""
raw = "Origin: http://example.com:8080"
headers = http_headers(raw)
self.assertEqual(headers["Origin"], "http://example.com:8080")
class TestKeys(unittest.TestCase):
def test_make_accept_rfc(self):
"""
Test ``make_accept()`` using the keys listed in the RFC for HyBi-07
through HyBi-10.
"""
key = "dGhlIHNhbXBsZSBub25jZQ=="
self.assertEqual(make_accept(key), "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=")
def test_make_accept_wikipedia(self):
"""
Test ``make_accept()`` using the keys listed on Wikipedia.
"""
key = "x3JJHMbDL1EzLkh9GBhXDw=="
self.assertEqual(make_accept(key), "HSmrc0sMlYUkAGmm5OPpG2HaGWk=")
class TestHyBi00(unittest.TestCase):
def test_is_hybi00(self):
headers = {
"Sec-WebSocket-Key1": "hurp",
"Sec-WebSocket-Key2": "derp",
}
self.assertTrue(is_hybi00(headers))
def test_is_hybi00_false(self):
headers = {
"Sec-WebSocket-Key1": "hurp",
}
self.assertFalse(is_hybi00(headers))
def test_complete_hybi00_wikipedia(self):
"""
Test complete_hybi00() using the keys listed on Wikipedia's WebSockets
page.
"""
headers = {
"Sec-WebSocket-Key1": "4 @1 46546xW%0l 1 5",
"Sec-WebSocket-Key2": "12998 5 Y3 1 .P00",
}
challenge = "^n:ds[4U"
self.assertEqual(complete_hybi00(headers, challenge),
six.b("8jKS'y:G*Co,Wxa-"))
def test_make_hybi00(self):
"""
HyBi-00 frames are really, *really* simple.
"""
self.assertEqual(make_hybi00_frame("Test!"), six.b("\x00Test!\xff"))
def test_parse_hybi00_single(self):
frame = six.b("\x00Test\xff")
frames, buf = parse_hybi00_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (NORMAL, six.b("Test")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi00_multiple(self):
frame = six.b("\x00Test\xff\x00Again\xff")
frames, buf = parse_hybi00_frames(frame)
self.assertEqual(len(frames), 2)
self.assertEqual(frames[0], (NORMAL, six.b("Test")))
self.assertEqual(frames[1], (NORMAL, six.b("Again")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi00_incomplete(self):
frame = six.b("\x00Test")
frames, buf = parse_hybi00_frames(frame)
self.assertEqual(len(frames), 0)
self.assertEqual(buf, six.b("\x00Test"))
def test_parse_hybi00_garbage(self):
frame = six.b("trash\x00Test\xff")
frames, buf = parse_hybi00_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (NORMAL, six.b("Test")))
self.assertEqual(buf, six.b(""))
def test_socketio_crashers(self):
"""
A series of snippets which crash other WebSockets implementations
(specifically, Socket.IO) are harmless to this implementation.
"""
frames = [
"""[{"length":1}]""",
"""{"messages":[{"length":1}]}""",
"hello",
"hello<script>alert(/xss/)</script>",
"hello<img src=x:x onerror=alert(/xss.2/)>",
"{",
"~m~EVJLFDJP~",
]
for frame in frames:
prepared = make_hybi00_frame(frame)
frames, buf = parse_hybi00_frames(prepared)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (NORMAL, frame.encode('utf-8')))
self.assertEqual(buf, six.b(""))
class TestHyBi07Helpers(unittest.TestCase):
"""
HyBi-07 is best understood as a large family of helper functions which
work together, somewhat dysfunctionally, to produce a mediocre
Thanksgiving every other year.
"""
def test_mask_noop(self):
key = six.b("\x00\x00\x00\x00")
self.assertEqual(mask(six.b("Test"), key), six.b("Test"))
def test_mask_noop_long(self):
key = six.b("\x00\x00\x00\x00")
self.assertEqual(mask(six.b("LongTest"), key), six.b("LongTest"))
def test_parse_hybi07_unmasked_text(self):
"""
From HyBi-10, 4.7.
"""
frame = six.b("\x81\x05Hello")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (NORMAL, six.b("Hello")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_masked_text(self):
"""
From HyBi-10, 4.7.
"""
frame = six.b("\x81\x857\xfa!=\x7f\x9fMQX")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (NORMAL, six.b("Hello")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_unmasked_text_fragments(self):
"""
We don't care about fragments. We are totally unfazed.
From HyBi-10, 4.7.
"""
frame = six.b("\x01\x03Hel\x80\x02lo")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 2)
self.assertEqual(frames[0], (NORMAL, six.b("Hel")))
self.assertEqual(frames[1], (NORMAL, six.b("lo")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_ping(self):
"""
From HyBi-10, 4.7.
"""
frame = six.b("\x89\x05Hello")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (PING, six.b("Hello")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_pong(self):
"""
From HyBi-10, 4.7.
"""
frame = six.b("\x8a\x05Hello")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (PONG, six.b("Hello")))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_close_empty(self):
"""
A HyBi-07 close packet may have no body. In that case, it should use
the generic error code 1000, and have no reason.
"""
frame = six.b("\x88\x00")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (CLOSE, (1000, six.b("No reason given"))))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_close_reason(self):
"""
A HyBi-07 close packet must have its first two bytes be a numeric
error code, and may optionally include trailing text explaining why
the connection was closed.
"""
frame = six.b("\x88\x0b\x03\xe8No reason")
frames, buf = parse_hybi07_frames(frame)
self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], (CLOSE, (1000, six.b("No reason"))))
self.assertEqual(buf, six.b(""))
def test_parse_hybi07_partial_no_length(self):
frame = six.b("\x81")
frames, buf = parse_hybi07_frames(frame)
self.assertFalse(frames)
self.assertEqual(buf, six.b("\x81"))
def test_parse_hybi07_partial_truncated_length_int(self):
frame = six.b("\x81\xfe")
frames, buf = parse_hybi07_frames(frame)
self.assertFalse(frames)
self.assertEqual(buf, six.b("\x81\xfe"))
def test_parse_hybi07_partial_truncated_length_double(self):
frame = six.b("\x81\xff")
frames, buf = parse_hybi07_frames(frame)
self.assertFalse(frames)
self.assertEqual(buf, six.b("\x81\xff"))
def test_parse_hybi07_partial_no_data(self):
frame = six.b("\x81\x05")
frames, buf = parse_hybi07_frames(frame)
self.assertFalse(frames)
self.assertEqual(buf, six.b("\x81\x05"))
def test_parse_hybi07_partial_truncated_data(self):
frame = six.b("\x81\x05Hel")
frames, buf = parse_hybi07_frames(frame)
self.assertFalse(frames)
self.assertEqual(buf, six.b("\x81\x05Hel"))