Clean up code

opendrop/__init__.py

@@ -18,15 +18,15 @@ along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
 import logging
-import platform
 import os
+import platform
 
 __version__ = '0.10.3'
 
 if platform.system() == 'Darwin':
-	dyld_path = os.environ.get('DYLD_LIBRARY_PATH', '')  # save old path
-	openssl_path = '/usr/local/opt/openssl/lib'
-	archive_path = '/usr/local/opt/libarchive/lib'
-	os.environ['DYLD_LIBRARY_PATH'] = '{}:{}:{}'.format(dyld_path, openssl_path, archive_path)
+    dyld_path = os.environ.get('DYLD_LIBRARY_PATH', '')  # save old path
+    openssl_path = '/usr/local/opt/openssl/lib'
+    archive_path = '/usr/local/opt/libarchive/lib'
+    os.environ['DYLD_LIBRARY_PATH'] = '{}:{}:{}'.format(dyld_path, openssl_path, archive_path)
 
 logger = logging.getLogger(__name__)

opendrop/cli.py

@@ -17,15 +17,13 @@ You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
-import time
-
-import ipaddress
-import logging
 import argparse
-import sys
 import json
+import logging
 import os
+import sys
 import threading
+import time
 
 from .client import AirDropBrowser, AirDropClient
 from .config import AirDropConfig, AirDropReceiverFlags
@@ -39,7 +37,6 @@ def main():
 
 
 class AirDropCli:
-
     def __init__(self, args):
         parser = argparse.ArgumentParser()
         parser.add_argument('action', choices=['receive', 'find', 'send'])
@@ -60,9 +57,12 @@ class AirDropCli:
 
         # TODO put emails and phone in canonical form (lower case, no '+' sign, etc.)
 
-        self.config = AirDropConfig(email=args.email, phone=args.phone,
-                                    computer_name=args.name, computer_model=args.model,
-                                    debug=args.debug, interface=args.interface)
+        self.config = AirDropConfig(email=args.email,
+                                    phone=args.phone,
+                                    computer_name=args.name,
+                                    computer_model=args.model,
+                                    debug=args.debug,
+                                    interface=args.interface)
         self.server = None
         self.client = None
         self.browser = None
@@ -106,7 +106,7 @@ class AirDropCli:
                 json.dump(self.discover, f)
 
     def _found_receiver(self, info):
-        thread = threading.Thread(target=self._send_discover, args=(info,))
+        thread = threading.Thread(target=self._send_discover, args=(info, ))
         thread.start()
 
     def _send_discover(self, info):
@@ -193,7 +193,7 @@ class AirDropCli:
         except IndexError:
             pass
         # (2) try 'id'
-        if len(self.receiver) is 12:
+        if len(self.receiver) == 12:
             for info in infos:
                 if info['id'] == self.receiver:
                     return info

opendrop/client.py

@@ -17,16 +17,17 @@ You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
-import fleep
-from http.client import HTTPSConnection
+import io
 import ipaddress
 import logging
 import os
-import io
-import libarchive
 import platform
 import plistlib
 import socket
+from http.client import HTTPSConnection
+
+import fleep
+import libarchive
 
 from .util import AirDropUtil, AbsArchiveWrite
 from .zeroconf import ServiceBrowser, Zeroconf, IPVersion
@@ -35,11 +36,10 @@ logger = logging.getLogger(__name__)
 
 
 class AirDropBrowser:
-
     def __init__(self, config):
         self.ip_addr = AirDropUtil.get_ip_for_interface(config.interface, ipv6=True)
         if self.ip_addr is None:
-            if config.interface is 'awdl0':
+            if config.interface == 'awdl0':
                 raise RuntimeError('Interface {} does not have an IPv6 address. '
                                    'Make sure that `owl` is running.'.format(config.interface))
             else:
@@ -80,7 +80,6 @@ class AirDropBrowser:
 
 
 class AirDropClient:
-
     def __init__(self, config, receiver):
         self.config = config
         self.receiver_host = receiver[0]
@@ -98,7 +97,8 @@ class AirDropClient:
                 _headers[key] = val
         if self.http_conn is None:
             # Use single connection
-            self.http_conn = HTTPSConnectionAWDL(self.receiver_host, self.receiver_port,
+            self.http_conn = HTTPSConnectionAWDL(self.receiver_host,
+                                                 self.receiver_port,
                                                  interface_name=self.config.interface,
                                                  context=self.config.get_ssl_context())
         self.http_conn.request('POST', url, body=body, headers=_headers)
@@ -161,6 +161,7 @@ class AirDropClient:
                     'ConvertMediaFormats': 0
                 }
                 yield file_entry
+
         ask_body['Files'] = [e for e in file_entries(file_path)]
         ask_body['Items'] = []
 
@@ -195,7 +196,7 @@ class AirDropClient:
 
     def _get_headers(self):
         """
-        Get the headers for requests sent 
+        Get the headers for requests sent
         """
         headers = {
             'Content-Type': 'application/octet-stream',
@@ -212,9 +213,17 @@ class HTTPSConnectionAWDL(HTTPSConnection):
     """
     This class allows to bind the HTTPConnection to a specific network interface
     """
-
-    def __init__(self, host, port=None, key_file=None, cert_file=None, timeout=None, source_address=None,
-                 *, context=None, check_hostname=None, interface_name=None):
+    def __init__(self,
+                 host,
+                 port=None,
+                 key_file=None,
+                 cert_file=None,
+                 timeout=None,
+                 source_address=None,
+                 *,
+                 context=None,
+                 check_hostname=None,
+                 interface_name=None):
 
         if interface_name is not None:
             if '%' not in host:
@@ -224,8 +233,13 @@ class HTTPSConnectionAWDL(HTTPSConnection):
         if timeout is None:
             timeout = socket.getdefaulttimeout()
 
-        super(HTTPSConnectionAWDL, self).__init__(host=host, port=port, key_file=key_file, cert_file=cert_file,
-                                                  timeout=timeout, source_address=source_address, context=context,
+        super(HTTPSConnectionAWDL, self).__init__(host=host,
+                                                  port=port,
+                                                  key_file=key_file,
+                                                  cert_file=cert_file,
+                                                  timeout=timeout,
+                                                  source_address=source_address,
+                                                  context=context,
                                                   check_hostname=check_hostname)
 
         self.interface_name = interface_name

opendrop/config.py

@@ -17,14 +17,15 @@ You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
-import os
 import logging
-from pkg_resources import resource_filename
+import os
+import random
 import socket
 import ssl
-import random
 import subprocess
 
+from pkg_resources import resource_filename
+
 logger = logging.getLogger(__name__)
 
 
@@ -45,10 +46,17 @@ class AirDropReceiverFlags:
 
 
 class AirDropConfig:
-
-    def __init__(self, host_name=None, computer_name=None, computer_model=None, server_port=8771,
-                 airdrop_dir='~/.opendrop', service_id=None,
-                 email=None, phone=None, debug=False, interface=None):
+    def __init__(self,
+                 host_name=None,
+                 computer_name=None,
+                 computer_model=None,
+                 server_port=8771,
+                 airdrop_dir='~/.opendrop',
+                 service_id=None,
+                 email=None,
+                 phone=None,
+                 debug=False,
+                 interface=None):
         self.airdrop_dir = os.path.expanduser(airdrop_dir)
 
         self.discovery_report = os.path.join(self.airdrop_dir, 'discover.last.json')
@@ -104,14 +112,18 @@ class AirDropConfig:
         logger.info('Create new self-signed certificate in {}'.format(self.key_dir))
         if not os.path.exists(self.key_dir):
             os.makedirs(self.key_dir)
-        subprocess.run(['openssl', 'req', '-newkey', 'rsa:2048', '-nodes', '-keyout', 'key.pem',
-                        '-x509', '-days', '365', '-out', 'certificate.pem',
-                        '-subj', '/CN={}'.format(self.computer_name)], cwd=self.key_dir,
-                       stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        subprocess.run([
+            'openssl', 'req', '-newkey', 'rsa:2048', '-nodes', '-keyout', 'key.pem', '-x509', '-days', '365', '-out',
+            'certificate.pem', '-subj', '/CN={}'.format(self.computer_name)
+        ],
+                       cwd=self.key_dir,
+                       stdout=subprocess.PIPE,
+                       stderr=subprocess.PIPE)
 
     def get_ssl_context(self):
-        sslctxt = ssl.SSLContext()
-        sslctxt.load_cert_chain(self.cert_file, keyfile=self.key_file)
-        sslctxt.load_verify_locations(cafile=self.root_ca_file)
-        sslctxt.verify_mode = ssl.CERT_NONE  # we accept self-signed certificates as does Apple
-        return sslctxt
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS)
+        ctx.options |= ssl.OP_NO_TLSv1  # TLSv1.0 is insecure
+        ctx.load_cert_chain(self.cert_file, keyfile=self.key_file)
+        ctx.load_verify_locations(cafile=self.root_ca_file)
+        ctx.verify_mode = ssl.CERT_NONE  # we accept self-signed certificates as does Apple
+        return ctx

opendrop/server.py

@@ -17,20 +17,20 @@ You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 import io
-
+import json
 import logging
 import platform
 import plistlib
 import socket
+import time
 from http.server import HTTPServer, BaseHTTPRequestHandler
-import json
+
 import libarchive
 import libarchive.extract
 import libarchive.read
-from .zeroconf import Zeroconf, ServiceInfo, IPVersion
-import time
 
 from .util import AirDropUtil
+from .zeroconf import Zeroconf, ServiceInfo, IPVersion
 
 logger = logging.getLogger(__name__)
 
@@ -39,7 +39,6 @@ class AirDropServer:
     """
     Announces an HTTPS AirDrop server in the local network via mDNS.
     """
-
     def __init__(self, config):
         self.config = config
 
@@ -50,7 +49,7 @@ class AirDropServer:
 
         self.ip_addr = AirDropUtil.get_ip_for_interface(self.config.interface, ipv6=True)
         if self.ip_addr is None:
-            if self.config.interface is 'awdl0':
+            if self.config.interface == 'awdl0':
                 raise RuntimeError('Interface {} does not have an IPv6 address. '
                                    'Make sure that `owl` is running.'.format(self.config.interface))
             else:
@@ -68,14 +67,13 @@ class AirDropServer:
         properties = self.get_properties()
         server = self.config.host_name + '.local.'
         service_name = self.config.service_id + '._airdrop._tcp.local.'
-        info = ServiceInfo(
-            '_airdrop._tcp.local.', service_name,
-            self.ip_addr.packed, self.config.port, 0, 0, properties, server)
+        info = ServiceInfo('_airdrop._tcp.local.', service_name, self.ip_addr.packed, self.config.port, 0, 0, properties,
+                           server)
         return info
 
     def start_service(self):
-        logger.info('Announcing service: host {}, address {}, port {}'.format(self.config.host_name,
-                                                                              self.ip_addr, self.config.port))
+        logger.info('Announcing service: host {}, address {}, port {}'.format(self.config.host_name, self.ip_addr,
+                                                                              self.config.port))
         self.zeroconf.register_service(self.service_info)
 
     def _init_server(self):
@@ -203,8 +201,7 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
 
         AirDropUtil.write_debug(self.config, post_data, 'receive_ask_request.plist')
 
-        ask_response = {'ReceiverModelName': self.config.computer_model,
-                        'ReceiverComputerName': self.config.computer_name}
+        ask_response = {'ReceiverModelName': self.config.computer_model, 'ReceiverComputerName': self.config.computer_name}
         ask_resp_binary = plistlib.dumps(ask_response, fmt=plistlib.FMT_BINARY)
 
         AirDropUtil.write_debug(self.config, ask_resp_binary, 'receive_ask_response.plist')
@@ -245,18 +242,18 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
                 self.total = 0
 
             def _next_chunk(self):
-                if self.chunk is None or len(self.chunk) is 0:
+                if self.chunk is None or len(self.chunk) == 0:
                     length = int(self.rfile.readline().rstrip(), 16)
                     self.chunk = self.rfile.read(length)
                     self.rfile.readline()  # strip trailing \n\r
 
             def readinto(self, buf):
                 self._next_chunk()
-                l = min(len(self.chunk), len(buf))
-                buf[:l] = self.chunk[:l]
-                self.chunk = self.chunk[l:]
-                self.total += l
-                return l
+                length = min(len(self.chunk), len(buf))
+                buf[:length] = self.chunk[:length]
+                self.chunk = self.chunk[length:]
+                self.total += length
+                return length
 
         def extract_stream(stream, flags=0):
             """

opendrop/util.py

@@ -19,13 +19,15 @@ along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import base64
 import datetime
+import hashlib
 import io
 import ipaddress
 import os
 import plistlib
-import hashlib
+
 import ifaddr
 from PIL import Image, ExifTags
+from ctypescrypto import cms, x509, pkey, oid
 from libarchive import ffi
 from libarchive.entry import new_archive_entry, ArchiveEntry
 from libarchive.ffi import (
@@ -39,7 +41,6 @@ from libarchive.ffi import (
     write_finish_entry,
 )
 from libarchive.write import ArchiveWrite, new_archive_read_disk
-from ctypescrypto import cms, x509, pkey, oid
 
 
 class AirDropUtil:
@@ -47,13 +48,12 @@ class AirDropUtil:
     This class contains a set of utility functions that support the opendrop implementation
     They have been moved, because the opendrop files tend to get too long
     """
-
     @staticmethod
     def get_uti_type(flp) -> str:
         """
-        Get the Apple conform UTI Type from a flp instance which has been used on the data which should be sent 
+        Get the Apple conform UTI Type from a flp instance which has been used on the data which should be sent
 
-        :param flp: fleep object 
+        :param flp: fleep object
         """
 
         # Default UTI Type
@@ -132,8 +132,7 @@ class AirDropUtil:
                 cert = x509.X509(sign_cert_file.read())
                 key = pkey.PKey(privkey=key_file.read())
                 # possibly need to add intermediate certs
-                cms_signed = cms.SignedData.create(record_data_plist, cert=cert, pkey=key, certs=None,
-                                                   flags=cms.Flags.PARTIAL)
+                cms_signed = cms.SignedData.create(record_data_plist, cert=cert, pkey=key, certs=None, flags=cms.Flags.PARTIAL)
                 signed_data = AirDropUtil.pem2der(cms_signed.pem())
 
         return signed_data
@@ -141,7 +140,7 @@ class AirDropUtil:
     @staticmethod
     def pem2der(s):
         """
-        Create DER Formatted bytes from a PEM Base64 String 
+        Create DER Formatted bytes from a PEM Base64 String
 
         :param s: PEM formatted string
         """
@@ -153,10 +152,10 @@ class AirDropUtil:
     @staticmethod
     def generate_file_icon(file_path):
         """
-        Generates a small and a big thumbnail of an image 
+        Generates a small and a big thumbnail of an image
         This will make it possible to preview the sent file
 
-        :param file_path: The path to the image 
+        :param file_path: The path to the image
         """
         im = Image.open(file_path)
 
@@ -177,10 +176,10 @@ class AirDropUtil:
         file_icon = imgByteArr.getvalue()
 
         # Small image
-        #im.thumbnail((64, 64), Image.ANTIALIAS)
-        #imgByteArr = io.BytesIO()
-        #im.save(imgByteArr, format='JPEG2000')
-        #small_file_icon = imgByteArr.getvalue()
+        # im.thumbnail((64, 64), Image.ANTIALIAS)
+        # imgByteArr = io.BytesIO()
+        # im.save(imgByteArr, format='JPEG2000')
+        # small_file_icon = imgByteArr.getvalue()
 
         return file_icon
 
@@ -193,7 +192,6 @@ class AirDropUtil:
         :param bool ipv6: Boolean indicating if the ipv6 address should be retrieved
         :return: IPv4Address or IPv6Address object or None
         """
-
         def get_interface_by_name(name):
             for interface in ifaddr.get_adapters():
                 if interface.name == name:
@@ -241,7 +239,7 @@ class AbsArchiveWrite(ArchiveWrite):
         with new_archive_entry() as entry_p:
             entry = ArchiveEntry(None, entry_p)
             with new_archive_read_disk(path) as read_p:
-                while 1:
+                while True:
                     r = read_next_header2(read_p, entry_p)
                     if r == ARCHIVE_EOF:
                         break
@@ -250,7 +248,7 @@ class AbsArchiveWrite(ArchiveWrite):
                     write_header(write_p, entry_p)
                     try:
                         with open(entry_sourcepath(entry_p), 'rb') as f:
-                            while 1:
+                            while True:
                                 data = f.read(block_size)
                                 if not data:
                                     break

setup.py

@@ -32,11 +32,8 @@ setup(
     ],
     keywords='cli',
     packages=find_packages(exclude=['docs']),
-    package_data={
-        'opendrop': ['certs/*.pem']
-    },
-    install_requires=['requests', 'fleep', 'ifaddr', 'Pillow',
-                      'requests_toolbelt', 'ctypescrypto', 'libarchive-c'],
+    package_data={'opendrop': ['certs/*.pem']},
+    install_requires=['requests', 'fleep', 'ifaddr', 'Pillow', 'requests_toolbelt', 'ctypescrypto', 'libarchive-c'],
     entry_points={
         'console_scripts': [
             'opendrop=opendrop.cli:main',