Bump version

Otherwise discovery would stop after entering key passphrase

.github/workflows/testpackage.yml

@@ -8,7 +8,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.6, 3.7, 3.8]
+        python-version: [3.6, 3.7, 3.8, 3.9]
 
     steps:
     - uses: actions/checkout@v1
@@ -22,14 +22,22 @@ jobs:
     - name: Install package
       run: |
         pip install -e .
-    - name: Check format with yapf
+    - name: Check format with isort
       run: |
-        pip install yapf
-        yapf . -r --diff
+        pip install isort
+        isort -c opendrop/**.py
+    - name: Check format with black
+      run: |
+        pip install black
+        black . --check --diff
     - name: Lint with flake8
       run: |
         pip install flake8
         flake8 . --count --show-source --statistics
+    - name: Lint with pylint
+      run: |
+        pip install pylint
+        pylint --rcfile=setup.cfg opendrop
     - name: Test with pytest
       run: |
         pip install pytest

Makefile

@@ -1,9 +1,9 @@
-.PHONY: ci checkformat lint test autoformat
+.PHONY: ci checkformat isort lint pylint test autoformat
 
 VENV=venv
 PYTHON=$(VENV)/bin/python3
 
-ci: checkformat lint test
+ci: isort checkformat lint pylint test
 
 $(VENV): $(VENV)/bin/activate
 
@@ -18,13 +18,20 @@ endif
 	touch $(VENV)/bin/activate
 
 checkformat: $(VENV)
-	$(PYTHON) -m yapf . -r --diff --exclude $(VENV)
+	$(PYTHON) -m black . --check --diff --exclude $(VENV)
 
 lint: $(VENV)	
 	$(PYTHON) -m flake8 . --count --show-source --statistics --exclude $(VENV)
 
+pylint: $(VENV)
+	$(PYTHON) -m pylint --rcfile=setup.cfg opendrop
+
+isort: $(VENV)
+	$(PYTHON) -m isort -c opendrop/**.py
+
 test: $(VENV)
 	$(PYTHON) -m pytest
 
 autoformat: $(VENV)
-	$(PYTHON) -m yapf . -r --in-place --exclude $(VENV)
+	$(PYTHON) -m isort opendrop/**.py
+	$(PYTHON) -m black . --exclude $(VENV)

README.md

@@ -1,8 +1,11 @@
 # OpenDrop: an Open Source AirDrop Implementation
 
-*OpenDrop* is a command-line tool that allows sharing files between devices directly over Wi-Fi. Its unique feature is that it is protocol-compatible with Apple AirDrop which allows to share files with Apple devices running iOS and macOS. 
-Currently (and probably also for the foreseeable future), OpenDrop only supports sending to Apple devices that are discoverable by *everybody* as the default *contacts only* mode requires [Apple-signed certificates](https://www.apple.com/certificateauthority/pdf/Apple_AAI_CPS_v6.1.pdf).
+[![Release](https://img.shields.io/pypi/v/opendrop?color=%23EC6500&label=release)](https://pypi.org/project/opendrop/)
+[![Language grade](https://img.shields.io/lgtm/grade/python/github/seemoo-lab/opendrop?label=code%20quality)](https://lgtm.com/projects/g/seemoo-lab/opendrop/context:python)
 
+*OpenDrop* is a command-line tool that allows sharing files between devices directly over Wi-Fi. Its unique feature is that it is protocol-compatible with Apple AirDrop which allows to share files with Apple devices running iOS and macOS. 
+~~Currently (and probably also for the foreseeable future), OpenDrop only supports sending to Apple devices that are discoverable by *everybody* as the default *contacts only* mode requires [Apple-signed certificates](https://www.apple.com/certificateauthority/pdf/Apple_AAI_CPS_v6.1.pdf).~~
+We support contacts-only devices by using extracted AirDrop credentials (keys and certificates) from macOS via our [keychain extractor](https://github.com/seemoo-lab/airdrop-keychain-extractor).
 
 ## Disclaimer
 
@@ -20,19 +23,19 @@ In addition, it requires Python >=3.6 as well as several libraries.
 As AirDrop exclusively runs over Apple Wireless Direct Link (AWDL), OpenDrop is only supported on macOS or on Linux systems running an open re-implementation of AWDL such as [OWL](https://github.com/seemoo-lab/owl).
 
 **Libraries.**
-OpenDrop relies on current versions of [OpenSSL](https://www.openssl.org) and [libarchive](https://www.libarchive.org).
-macOS ships with rather old versions of the two, so you will need to install newer version, for example, via [Homebrew](https://brew.sh):
+OpenDrop relies on a current version of [libarchive](https://www.libarchive.org).
+macOS ships with a rather old version, so you will need to install a newer version, for example, via [Homebrew](https://brew.sh):
 ```bash
-brew install libarchive openssl
+brew install libarchive
 ```
-OpenDrop automatically sets `DYLD_LIBRARY_PATH` to look for the Homebrew versions. You may need to update the variable yourself if you install the libraries differently.
+OpenDrop automatically sets `DYLD_LIBRARY_PATH` to look for the Homebrew version. You may need to update the variable yourself if you install the libraries differently.
 
 Linux distributions should ship with more up-to-date versions, so this won't be necessary.
 
 
 ## Installation 
 
-Installation of the Python package [release](https://pypi.org/project/opendrop/) is straight forward using `pip3`:
+Installation of the Python package [release](https://pypi.org/project/opendrop/) is straightforward using `pip3`:
 ```
 pip3 install opendrop
 ```
@@ -55,7 +58,7 @@ Sending a file is typically a two-step procedure. You first discover devices in
 Stop the process once you have found the receiver.
 ```
 $ opendrop find
-Looking for receivers. Press enter to stop ...
+Looking for receivers. Press Ctrl+C to stop ...
 Found  index 0  ID eccb2f2dcfe7  name John’s iPhone
 Found  index 1  ID e63138ac6ba8  name Jane’s MacBook Pro
 ```
@@ -89,7 +92,7 @@ OpenDrop is the result of a research project and, thus, has several limitations
 * *Sending multiple files.* Apple AirDrop supports sending multiple files at once, OpenDrop does not (would require adding more files to the archive, modify HTTP /Ask request, etc.).
 
 
-## Related Papers
+## Our Papers
 
 * Milan Stute, Sashank Narain, Alex Mariotto, Alexander Heinrich, David Kreitschmann, Guevara Noubir, and Matthias Hollick. **A Billion Open Interfaces for Eve and Mallory: MitM, DoS, and Tracking Attacks on iOS and macOS Through Apple Wireless Direct Link.** *28th USENIX Security Symposium (USENIX Security ’19)*, August 14–16, 2019, Santa Clara, CA, USA. [Link](https://www.usenix.org/conference/usenixsecurity19/presentation/stute)
 

opendrop/__init__.py

@@ -21,12 +21,11 @@ import logging
 import os
 import platform
 
-__version__ = '0.11.0'
+__version__ = "0.12.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)
+if platform.system() == "Darwin":
+    dyld_path = os.environ.get("DYLD_LIBRARY_PATH", "")  # save old path
+    archive_path = "/usr/local/opt/libarchive/lib"
+    os.environ["DYLD_LIBRARY_PATH"] = f"{dyld_path}:{archive_path}"
 
 logger = logging.getLogger(__name__)

opendrop/cli.py

@@ -39,30 +39,51 @@ def main():
 class AirDropCli:
     def __init__(self, args):
         parser = argparse.ArgumentParser()
-        parser.add_argument('action', choices=['receive', 'find', 'send'])
-        parser.add_argument('-f', '--file', help='File to be sent')
-        parser.add_argument('-r', '--receiver', help='Peer to send file to (can be index, ID, or hostname)')
-        parser.add_argument('-e', '--email', nargs='*', help='User\'s email addresses (currently unused)')
-        parser.add_argument('-p', '--phone', nargs='*', help='User\'s phone numbers (currently unused)')
-        parser.add_argument('-n', '--name', help='Computer name (displayed in sharing pane)')
-        parser.add_argument('-m', '--model', help='Computer model (displayed in sharing pane)')
-        parser.add_argument('-d', '--debug', help='Enable debug mode', action='store_true')
-        parser.add_argument('-i', '--interface', help='Which AWDL interface to use', default='awdl0')
+        parser.add_argument("action", choices=["receive", "find", "send"])
+        parser.add_argument("-f", "--file", help="File to be sent")
+        parser.add_argument(
+            "-r",
+            "--receiver",
+            help="Peer to send file to (can be index, ID, or hostname)",
+        )
+        parser.add_argument(
+            "-e", "--email", nargs="*", help="User's email addresses (currently unused)"
+        )
+        parser.add_argument(
+            "-p", "--phone", nargs="*", help="User's phone numbers (currently unused)"
+        )
+        parser.add_argument(
+            "-n", "--name", help="Computer name (displayed in sharing pane)"
+        )
+        parser.add_argument(
+            "-m", "--model", help="Computer model (displayed in sharing pane)"
+        )
+        parser.add_argument(
+            "-d", "--debug", help="Enable debug mode", action="store_true"
+        )
+        parser.add_argument(
+            "-i", "--interface", help="Which AWDL interface to use", default="awdl0"
+        )
         args = parser.parse_args(args)
 
         if args.debug:
-            logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)-8s %(name)s: %(message)s')
+            logging.basicConfig(
+                level=logging.DEBUG,
+                format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
+            )
         else:
-            logging.basicConfig(level=logging.INFO, format='%(message)s')
+            logging.basicConfig(level=logging.INFO, format="%(message)s")
 
         # 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
@@ -71,18 +92,18 @@ class AirDropCli:
         self.lock = threading.Lock()
 
         try:
-            if args.action == 'receive':
+            if args.action == "receive":
                 self.receive()
-            elif args.action == 'find':
+            elif args.action == "find":
                 self.find()
             else:  # args.action == 'send'
                 if args.file is None:
-                    parser.error('Need -f,--file when using send')
+                    parser.error("Need -f,--file when using send")
                 if not os.path.isfile(args.file):
-                    parser.error('File in -f,--file not found')
+                    parser.error("File in -f,--file not found")
                 self.file = args.file
                 if args.receiver is None:
-                    parser.error('Need -r,--receiver when using send')
+                    parser.error("Need -r,--receiver when using send")
                 self.receiver = args.receiver
                 self.send()
         except KeyboardInterrupt:
@@ -92,36 +113,36 @@ class AirDropCli:
                 self.server.stop()
 
     def find(self):
-        logger.info('Looking for receivers. Press enter to stop ...')
+        logger.info("Looking for receivers. Press Ctrl+C to stop ...")
         self.browser = AirDropBrowser(self.config)
         self.browser.start(callback_add=self._found_receiver)
         try:
-            input()
+            threading.Event().wait()
         except KeyboardInterrupt:
             pass
         finally:
             self.browser.stop()
-            logger.debug('Save discovery results to {}'.format(self.config.discovery_report))
-            with open(self.config.discovery_report, 'w') as f:
+            logger.debug(f"Save discovery results to {self.config.discovery_report}")
+            with open(self.config.discovery_report, "w") as f:
                 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):
         try:
             address = info.parsed_addresses()[0]  # there should only be one address
         except IndexError:
-            logger.warn('Ignoring receiver with missing address {}'.format(info))
+            logger.warning(f"Ignoring receiver with missing address {info}")
             return
-        id = info.name.split('.')[0]
+        identifier = info.name.split(".")[0]
         hostname = info.server
         port = int(info.port)
-        logger.debug('AirDrop service found: {}, {}:{}, ID {}'.format(hostname, address, port, id))
+        logger.debug(f"AirDrop service found: {hostname}, {address}:{port}, ID {id}")
         client = AirDropClient(self.config, (address, int(port)))
         try:
-            flags = int(info.properties[b'flags'])
+            flags = int(info.properties[b"flags"])
         except KeyError:
             # TODO in some cases, `flags` are not set in service info; for now we'll try anyway
             flags = AirDropReceiverFlags.SUPPORTS_DISCOVER_MAYBE
@@ -137,19 +158,19 @@ class AirDropCli:
 
         index = len(self.discover)
         node_info = {
-            'name': receiver_name,
-            'address': address,
-            'port': port,
-            'id': id,
-            'flags': flags,
-            'discoverable': discoverable,
+            "name": receiver_name,
+            "address": address,
+            "port": port,
+            "id": identifier,
+            "flags": flags,
+            "discoverable": discoverable,
         }
         self.lock.acquire()
         self.discover.append(node_info)
         if discoverable:
-            logger.info('Found  index {}  ID {}  name {}'.format(index, id, receiver_name))
+            logger.info(f"Found  index {index}  ID {identifier}  name {receiver_name}")
         else:
-            logger.debug('Receiver ID {} is not discoverable'.format(id))
+            logger.debug(f"Receiver ID {identifier} is not discoverable")
         self.lock.release()
 
     def receive(self):
@@ -161,26 +182,28 @@ class AirDropCli:
         info = self._get_receiver_info()
         if info is None:
             return
-        self.client = AirDropClient(self.config, (info['address'], info['port']))
-        logger.info('Asking receiver to accept ...')
+        self.client = AirDropClient(self.config, (info["address"], info["port"]))
+        logger.info("Asking receiver to accept ...")
         if not self.client.send_ask(self.file):
-            logger.warning('Receiver declined')
+            logger.warning("Receiver declined")
             return
-        logger.info('Receiver accepted')
-        logger.info('Uploading file ...')
+        logger.info("Receiver accepted")
+        logger.info("Uploading file ...")
         if not self.client.send_upload(self.file):
-            logger.warning('Uploading has failed')
+            logger.warning("Uploading has failed")
             return
-        logger.info('Uploading has been successful')
+        logger.info("Uploading has been successful")
 
     def _get_receiver_info(self):
         if not os.path.exists(self.config.discovery_report):
-            logger.error('No discovery report exists, please run \'opendrop find\' first')
+            logger.error("No discovery report exists, please run 'opendrop find' first")
             return None
         age = time.time() - os.path.getmtime(self.config.discovery_report)
         if age > 60:  # warn if report is older than a minute
-            logger.warning('Old discovery report (%.1f seconds), consider running \'opendrop find\' again', age)
-        with open(self.config.discovery_report, 'r') as f:
+            logger.warning(
+                f"Old discovery report ({age:.1f} seconds), consider running 'opendrop find' again"
+            )
+        with open(self.config.discovery_report, "r") as f:
             infos = json.load(f)
 
         # (1) try 'index'
@@ -194,12 +217,14 @@ class AirDropCli:
         # (2) try 'id'
         if len(self.receiver) == 12:
             for info in infos:
-                if info['id'] == self.receiver:
+                if info["id"] == self.receiver:
                     return info
         # (3) try hostname
         for info in infos:
-            if info['name'] == self.receiver:
+            if info["name"] == self.receiver:
                 return info
         # (fail)
-        logger.error('Receiver does not exist (check -r,--receiver format or try \'opendrop find\' again')
+        logger.error(
+            "Receiver does not exist (check -r,--receiver format or try 'opendrop find' again"
+        )
         return None

opendrop/client.py

@@ -28,9 +28,9 @@ from http.client import HTTPSConnection
 
 import fleep
 import libarchive
+from zeroconf import IPVersion, ServiceBrowser, Zeroconf
 
-from .util import AirDropUtil, AbsArchiveWrite
-from zeroconf import ServiceBrowser, Zeroconf, IPVersion
+from .util import AbsArchiveWrite, AirDropUtil
 
 logger = logging.getLogger(__name__)
 
@@ -39,15 +39,20 @@ 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 == 'awdl0':
-                raise RuntimeError('Interface {} does not have an IPv6 address. '
-                                   'Make sure that `owl` is running.'.format(config.interface))
+            if config.interface == "awdl0":
+                raise RuntimeError(
+                    f"Interface {config.interface} does not have an IPv6 address. Make sure that `owl` is running."
+                )
             else:
-                raise RuntimeError('Interface {} does not have an IPv6 address'.format(config.interface))
+                raise RuntimeError(
+                    f"Interface {config.interface} does not have an IPv6 address"
+                )
 
-        self.zeroconf = Zeroconf(interfaces=[str(self.ip_addr)],
-                                 ip_version=IPVersion.V6Only,
-                                 apple_p2p=platform.system() == 'Darwin')
+        self.zeroconf = Zeroconf(
+            interfaces=[str(self.ip_addr)],
+            ip_version=IPVersion.V6Only,
+            apple_p2p=platform.system() == "Darwin",
+        )
 
         self.callback_add = None
         self.callback_remove = None
@@ -61,22 +66,22 @@ class AirDropBrowser:
             return  # already started
         self.callback_add = callback_add
         self.callback_remove = callback_remove
-        self.browser = ServiceBrowser(self.zeroconf, '_airdrop._tcp.local.', self)
+        self.browser = ServiceBrowser(self.zeroconf, "_airdrop._tcp.local.", self)
 
     def stop(self):
         self.browser.cancel()
         self.browser = None
         self.zeroconf.close()
 
-    def add_service(self, zeroconf, type, name):
-        info = zeroconf.get_service_info(type, name)
-        logger.debug('Add service {}'.format(name))
+    def add_service(self, zeroconf, service_type, name):
+        info = zeroconf.get_service_info(service_type, name)
+        logger.debug(f"Add service {name}")
         if self.callback_add is not None:
             self.callback_add(info)
 
-    def remove_service(self, zeroconf, type, name):
-        info = zeroconf.get_service_info(type, name)
-        logger.debug('Remove service {}'.format(name))
+    def remove_service(self, zeroconf, service_type, name):
+        info = zeroconf.get_service_info(service_type, name)
+        logger.debug(f"Remove service {name}")
         if self.callback_remove is not None:
             self.callback_remove(info)
 
@@ -89,9 +94,11 @@ class AirDropClient:
         self.http_conn = None
 
     def send_POST(self, url, body, headers=None):
-        logger.debug('Send {} request'.format(url))
+        logger.debug(f"Send {url} request")
 
-        AirDropUtil.write_debug(self.config, body, 'send_{}_request.plist'.format(url.lower().strip('/')))
+        AirDropUtil.write_debug(
+            self.config, body, f"send_{url.lower().strip('/')}_request.plist"
+        )
 
         _headers = self._get_headers()
         if headers is not None:
@@ -99,76 +106,86 @@ class AirDropClient:
                 _headers[key] = val
         if self.http_conn is None:
             # Use single connection
-            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)
+            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)
         http_resp = self.http_conn.getresponse()
 
         response_bytes = http_resp.read()
-        AirDropUtil.write_debug(self.config, response_bytes, 'send_{}_response.plist'.format(url.lower().strip('/')))
+        AirDropUtil.write_debug(
+            self.config,
+            response_bytes,
+            f"send_{url.lower().strip('/')}_response.plist",
+        )
 
         if http_resp.status != 200:
             status = False
-            logger.debug('{} request failed: {}'.format(url, http_resp.status))
+            logger.debug(f"{url} request failed: {http_resp.status}")
         else:
             status = True
-            logger.debug('{} request successful'.format(url))
+            logger.debug(f"{url} request successful")
         return status, response_bytes
 
     def send_discover(self):
         discover_body = {}
         if self.config.record_data:
-            discover_body['SenderRecordData'] = self.config.record_data
+            discover_body["SenderRecordData"] = self.config.record_data
 
-        discover_plist_binary = plistlib.dumps(discover_body, fmt=plistlib.FMT_BINARY)
-        success, response_bytes = self.send_POST('/Discover', discover_plist_binary)
+        discover_plist_binary = plistlib.dumps(
+            discover_body, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
+        )
+        _, response_bytes = self.send_POST("/Discover", discover_plist_binary)
         response = plistlib.loads(response_bytes)
 
         # if name is returned, then receiver is discoverable
-        return response.get('ReceiverComputerName')
+        return response.get("ReceiverComputerName")
 
     def send_ask(self, file_path, icon=None):
         ask_body = {
-            'SenderComputerName': self.config.computer_name,
-            'BundleID': 'com.apple.finder',
-            'SenderModelName': self.config.computer_model,
-            'SenderID': self.config.service_id,
-            'ConvertMediaFormats': False,
+            "SenderComputerName": self.config.computer_name,
+            "BundleID": "com.apple.finder",
+            "SenderModelName": self.config.computer_model,
+            "SenderID": self.config.service_id,
+            "ConvertMediaFormats": False,
         }
         if self.config.record_data:
-            ask_body['SenderRecordData'] = self.config.record_data
+            ask_body["SenderRecordData"] = self.config.record_data
 
         if isinstance(file_path, str):
             file_path = [file_path]
 
         # generate icon for first file
-        with open(file_path[0], 'rb') as f:
+        with open(file_path[0], "rb") as f:
             file_header = f.read(128)
             flp = fleep.get(file_header)
-            if not icon and len(flp.mime) > 0 and 'image' in flp.mime[0]:
+            if not icon and len(flp.mime) > 0 and "image" in flp.mime[0]:
                 icon = AirDropUtil.generate_file_icon(f.name)
         if icon:
-            ask_body['FileIcon'] = icon
+            ask_body["FileIcon"] = icon
 
         def file_entries(files):
             for file in files:
                 file_name = os.path.basename(file)
                 file_entry = {
-                    'FileName': file_name,
-                    'FileType': AirDropUtil.get_uti_type(flp),
-                    'FileBomPath': os.path.join('.', file_name),
-                    'FileIsDirectory': os.path.isdir(file_name),
-                    'ConvertMediaFormats': 0
+                    "FileName": file_name,
+                    "FileType": AirDropUtil.get_uti_type(flp),
+                    "FileBomPath": os.path.join(".", file_name),
+                    "FileIsDirectory": os.path.isdir(file_name),
+                    "ConvertMediaFormats": 0,
                 }
                 yield file_entry
 
-        ask_body['Files'] = [e for e in file_entries(file_path)]
-        ask_body['Items'] = []
+        ask_body["Files"] = [e for e in file_entries(file_path)]
+        ask_body["Items"] = []
 
-        ask_binary = plistlib.dumps(ask_body, fmt=plistlib.FMT_BINARY)
-        success, _ = self.send_POST('/Ask', ask_binary)
+        ask_binary = plistlib.dumps(
+            ask_body, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
+        )
+        success, _ = self.send_POST("/Ask", ask_binary)
 
         return success
 
@@ -177,20 +194,24 @@ class AirDropClient:
         Send a file to a receiver.
         """
         headers = {
-            'Content-Type': 'application/x-cpio',
+            "Content-Type": "application/x-cpio",
         }
 
         # Create archive in memory ...
         stream = io.BytesIO()
-        with libarchive.custom_writer(stream.write, 'cpio', filter_name='gzip',
-                                      archive_write_class=AbsArchiveWrite) as archive:
+        with libarchive.custom_writer(
+            stream.write,
+            "cpio",
+            filter_name="gzip",
+            archive_write_class=AbsArchiveWrite,
+        ) as archive:
             for f in [file_path]:
                 ff = os.path.basename(f)
-                archive.add_abs_file(f, os.path.join('.', ff))
+                archive.add_abs_file(f, os.path.join(".", ff))
         stream.seek(0)
 
         # ... then send in chunked mode
-        success, _ = self.send_POST('/Upload', stream, headers=headers)
+        success, _ = self.send_POST("/Upload", stream, headers=headers)
 
         # TODO better: write archive chunk whenever send_POST does a read to avoid having the whole archive in memory
 
@@ -201,12 +222,12 @@ class AirDropClient:
         Get the headers for requests sent
         """
         headers = {
-            'Content-Type': 'application/octet-stream',
-            'Connection': 'keep-alive',
-            'Accept': '*/*',
-            'User-Agent': 'AirDrop/1.0',
-            'Accept-Language': 'en-us',
-            'Accept-Encoding': 'br, gzip, deflate'
+            "Content-Type": "application/octet-stream",
+            "Connection": "keep-alive",
+            "Accept": "*/*",
+            "User-Agent": "AirDrop/1.0",
+            "Accept-Language": "en-us",
+            "Accept-Encoding": "br, gzip, deflate",
         }
         return headers
 
@@ -215,39 +236,46 @@ 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:
+            if "%" not in host:
                 if isinstance(ipaddress.ip_address(host), ipaddress.IPv6Address):
-                    host = host + '%' + interface_name
+                    host = host + "%" + interface_name
 
         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,
-                                                  check_hostname=check_hostname)
+        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
         self._create_connection = self.create_connection_awdl
 
-    def create_connection_awdl(self, address, timeout=socket.getdefaulttimeout(), source_address=None):
+    def create_connection_awdl(
+        self, address, timeout=socket.getdefaulttimeout(), source_address=None
+    ):
         """Connect to *address* and return the socket object.
 
         Convenience function.  Connect to *address* (a 2-tuple ``(host,
@@ -263,13 +291,13 @@ class HTTPSConnectionAWDL(HTTPSConnection):
         host, port = address
         err = None
         for res in socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM):
-            af, socktype, proto, canonname, sa = res
+            af, socktype, proto, _, sa = res
             sock = None
             try:
                 sock = socket.socket(af, socktype, proto)
                 if timeout is not socket.getdefaulttimeout():
                     sock.settimeout(timeout)
-                if self.interface_name == 'awdl0' and platform.system() == 'Darwin':
+                if self.interface_name == "awdl0" and platform.system() == "Darwin":
                     sock.setsockopt(socket.SOL_SOCKET, 0x1104, 1)
                 if source_address:
                     sock.bind(source_address)
@@ -285,4 +313,4 @@ class HTTPSConnectionAWDL(HTTPSConnection):
         if err is not None:
             raise err
         else:
-            raise socket.error('getaddrinfo returns an empty list')
+            raise socket.error("getaddrinfo returns an empty list")

opendrop/config.py

@@ -34,7 +34,9 @@ class AirDropReceiverFlags:
     Recovered from sharingd`receiverSupportsX methods.
     A valid node needs to either have SUPPORTS_PIPELINING or SUPPORTS_MIXED_TYPES
     according to sharingd`[SDBonjourBrowser removeInvalidNodes:].
+    Default flags on macOS: 0x3fb according to sharingd`[SDRapportBrowser defaultSFNodeFlags]
     """
+
     SUPPORTS_URL = 0x01
     SUPPORTS_DVZIP = 0x02
     SUPPORTS_PIPELINING = 0x04
@@ -42,24 +44,30 @@ class AirDropReceiverFlags:
     SUPPORTS_UNKNOWN1 = 0x10
     SUPPORTS_UNKNOWN2 = 0x20
     SUPPORTS_IRIS = 0x40
-    SUPPORTS_DISCOVER_MAYBE = 0x80  # Probably indicates that server supports /Discover URL
+    SUPPORTS_DISCOVER_MAYBE = (
+        0x80  # Probably indicates that server supports /Discover URL
+    )
+    SUPPORTS_UNKNOWN3 = 0x100
+    SUPPORTS_ASSET_BUNDLE = 0x200
 
 
 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')
+        self.discovery_report = os.path.join(self.airdrop_dir, "discover.last.json")
 
         if host_name is None:
             host_name = socket.gethostname()
@@ -68,19 +76,19 @@ class AirDropConfig:
             computer_name = host_name
         self.computer_name = computer_name
         if computer_model is None:
-            computer_model = 'OpenDrop'
+            computer_model = "OpenDrop"
         self.computer_model = computer_model
         self.port = server_port
 
         if service_id is None:
-            service_id = '{0:0{1}x}'.format(random.randint(0, 0xffffffffffff), 12)  # random 6-byte string in base16
+            service_id = f"{random.randint(0, 0xFFFFFFFFFFFF):012x}"  # random 6-byte string in base16
         self.service_id = service_id
 
         self.debug = debug
-        self.debug_dir = os.path.join(self.airdrop_dir, 'debug')
+        self.debug_dir = os.path.join(self.airdrop_dir, "debug")
 
         if interface is None:
-            interface = 'awdl0'
+            interface = "awdl0"
         self.interface = interface
 
         if email is None:
@@ -91,39 +99,70 @@ class AirDropConfig:
         self.phone = phone
 
         # Bare minimum, we currently do not support anything else
-        self.flags = AirDropReceiverFlags.SUPPORTS_MIXED_TYPES | AirDropReceiverFlags.SUPPORTS_DISCOVER_MAYBE
+        self.flags = (
+            AirDropReceiverFlags.SUPPORTS_MIXED_TYPES
+            | AirDropReceiverFlags.SUPPORTS_DISCOVER_MAYBE
+        )
 
-        self.root_ca_file = resource_filename('opendrop', 'certs/apple_root_ca.pem')
+        self.root_ca_file = resource_filename("opendrop", "certs/apple_root_ca.pem")
         if not os.path.exists(self.root_ca_file):
-            raise FileNotFoundError('Need Apple root CA certificate: {}'.format(self.root_ca_file))
+            raise FileNotFoundError(
+                f"Need Apple root CA certificate: {self.root_ca_file}"
+            )
 
-        self.key_dir = os.path.join(self.airdrop_dir, 'keys')
-        self.cert_file = os.path.join(self.key_dir, 'certificate.pem')
-        self.key_file = os.path.join(self.key_dir, 'key.pem')
+        self.key_dir = os.path.join(self.airdrop_dir, "keys")
+        self.cert_file = os.path.join(self.key_dir, "certificate.pem")
+        self.key_file = os.path.join(self.key_dir, "key.pem")
 
         if not os.path.exists(self.cert_file) or not os.path.exists(self.key_file):
-            logger.info('Key file or certificate does not exist')
+            logger.info("Key file or certificate does not exist")
             self.create_default_key()
 
-        # TODO extract record data from a sample exchange
+        self.record_file = os.path.join(self.key_dir, "validation_record.cms")
         self.record_data = None
+        if os.path.exists(self.record_file):
+            logger.debug("Using provided Apple ID Validation Record")
+            with open(self.record_file, "rb") as f:
+                self.record_data = f.read()
+        else:
+            logger.debug("No Apple ID Validation Record found")
 
     def create_default_key(self):
-        logger.info('Create new self-signed certificate in {}'.format(self.key_dir))
+        logger.info(f"Create new self-signed certificate in {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",
+                f"/CN={self.computer_name}",
+            ],
+            cwd=self.key_dir,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            check=True,
+        )
 
     def get_ssl_context(self):
-        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS)
+
+        ctx = ssl.SSLContext(  # lgtm[py/insecure-protocol], TODO see https://github.com/Semmle/ql/issues/2554
+            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
+        ctx.verify_mode = (
+            ssl.CERT_NONE
+        )  # we accept self-signed certificates as does Apple
         return ctx

opendrop/server.py

@@ -23,14 +23,14 @@ import platform
 import plistlib
 import socket
 import time
-from http.server import HTTPServer, BaseHTTPRequestHandler
+from http.server import BaseHTTPRequestHandler, HTTPServer
 
 import libarchive
 import libarchive.extract
 import libarchive.read
+from zeroconf import IPVersion, ServiceInfo, Zeroconf
 
 from .util import AirDropUtil
-from zeroconf import Zeroconf, ServiceInfo, IPVersion
 
 logger = logging.getLogger(__name__)
 
@@ -39,47 +39,58 @@ class AirDropServer:
     """
     Announces an HTTPS AirDrop server in the local network via mDNS.
     """
+
     def __init__(self, config):
         self.config = config
 
         # Use IPv6
-        self.serveraddress = ('::', self.config.port)
+        self.serveraddress = ("::", self.config.port)
         self.ServerClass = HTTPServerV6
         self.ServerClass.allow_reuse_address = False
 
-        self.ip_addr = AirDropUtil.get_ip_for_interface(self.config.interface, ipv6=True)
+        self.ip_addr = AirDropUtil.get_ip_for_interface(
+            self.config.interface, ipv6=True
+        )
         if self.ip_addr is None:
-            if self.config.interface == 'awdl0':
-                raise RuntimeError('Interface {} does not have an IPv6 address. '
-                                   'Make sure that `owl` is running.'.format(self.config.interface))
+            if self.config.interface == "awdl0":
+                raise RuntimeError(
+                    f"Interface {self.config.interface} does not have an IPv6 address. Make sure that `owl` is running."
+                )
             else:
-                raise RuntimeError('Interface {} does not have an IPv6 address'.format(self.config.interface))
+                raise RuntimeError(
+                    f"Interface {self.config.interface} does not have an IPv6 address"
+                )
 
         self.Handler = AirDropServerHandler
         self.Handler.config = self.config
 
-        self.zeroconf = Zeroconf(interfaces=[str(self.ip_addr)],
-                                 ip_version=IPVersion.V6Only,
-                                 apple_p2p=platform.system() == 'Darwin')
+        self.zeroconf = Zeroconf(
+            interfaces=[str(self.ip_addr)],
+            ip_version=IPVersion.V6Only,
+            apple_p2p=platform.system() == "Darwin",
+        )
 
         self.http_server = self._init_server()
         self.service_info = self._init_service()
 
     def _init_service(self):
         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,
-                           port=self.config.port,
-                           properties=properties,
-                           server=server,
-                           addresses=[self.ip_addr.packed])
+        server = self.config.host_name + ".local."
+        service_name = self.config.service_id + "._airdrop._tcp.local."
+        info = ServiceInfo(
+            "_airdrop._tcp.local.",
+            service_name,
+            port=self.config.port,
+            properties=properties,
+            server=server,
+            addresses=[self.ip_addr.packed],
+        )
         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(
+            f"Announcing service: host {self.config.host_name}, address {self.ip_addr}, port {self.config.port}"
+        )
         self.zeroconf.register_service(self.service_info)
 
     def _init_server(self):
@@ -92,15 +103,17 @@ class AirDropServer:
             httpd = self.ServerClass(self.serveraddress, self.Handler)
 
         # Adapt socket for awdl0
-        if self.config.interface == 'awdl0' and platform.system() == 'Darwin':
+        if self.config.interface == "awdl0" and platform.system() == "Darwin":
             httpd.socket.setsockopt(socket.SOL_SOCKET, 0x1104, 1)
 
-        httpd.socket = self.config.get_ssl_context().wrap_socket(sock=httpd.socket, server_side=True)
+        httpd.socket = self.config.get_ssl_context().wrap_socket(
+            sock=httpd.socket, server_side=True
+        )
 
         return httpd
 
     def start_server(self):
-        logger.info('Starting HTTPS server')
+        logger.info("Starting HTTPS server")
         self.http_server.serve_forever()
 
     def stop(self):
@@ -108,7 +121,7 @@ class AirDropServer:
         self.http_server.shutdown()
 
     def get_properties(self):
-        properties = {b'flags': str(self.config.flags).encode('utf-8')}
+        properties = {b"flags": str(self.config.flags).encode("utf-8")}
         return properties
 
 
@@ -120,7 +133,8 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
     """
     Server which responds to AirDrop HTTP POST requests
     """
-    protocol_version = 'HTTP/1.1'
+
+    protocol_version = "HTTP/1.1"
     config = None
 
     def _set_response(self, content_length):
@@ -128,7 +142,7 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
         Setting the default values for a successful response
         """
         self.send_response(200)
-        self.send_header('Content-Length', content_length)
+        self.send_header("Content-Length", content_length)
         self.end_headers()
 
     def do_HEAD(self):
@@ -136,27 +150,29 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
         Answer head requests
         """
         self.send_response(200)
-        self.send_header('Content-type', 'text/html')
+        self.send_header("Content-type", "text/html")
         self.end_headers()
 
     def do_GET(self):
         """
         Answer get requests
         """
-        logger.debug('GET request at {}'.format(self.path))
-        body = '\n'.encode('utf-8')
+        logger.debug(f"GET request at {self.path}")
+        body = "\n".encode("utf-8")
         self._set_response(len(body))
         self.wfile.write(body)
 
     def handle_discover(self):
-        content_length = int(self.headers['Content-Length'])
+        content_length = int(self.headers["Content-Length"])
         post_data = self.rfile.read(content_length)
 
-        AirDropUtil.write_debug(self.config, post_data, 'receive_discover_request.plist')
+        AirDropUtil.write_debug(
+            self.config, post_data, "receive_discover_request.plist"
+        )
 
         # sample media capabilities as recorded from macOS 10.13.3
         media_capabilities = {
-            'Version': 1,
+            "Version": 1,
             # don't advertise any codec/container support so we receive legacy file formats (JPEG instead of HEIF, etc.)
             # 'Codecs': {
             #     'hvc1': {
@@ -184,59 +200,72 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
             # }
         }
         media_capabilities_json = json.JSONEncoder().encode(media_capabilities)
-        media_capabilities_binary = media_capabilities_json.encode('utf-8')
+        media_capabilities_binary = media_capabilities_json.encode("utf-8")
         discover_answer = {
-            'ReceiverMediaCapabilities': media_capabilities_binary,
-            'ReceiverComputerName': self.config.computer_name,
-            'ReceiverModelName': self.config.computer_model,
+            "ReceiverMediaCapabilities": media_capabilities_binary,
+            "ReceiverComputerName": self.config.computer_name,
+            "ReceiverModelName": self.config.computer_model,
         }
         if self.config.record_data:
-            discover_answer['ReceiverRecordData'] = self.config.record_data
+            discover_answer["ReceiverRecordData"] = self.config.record_data
 
-        discover_answer_binary = plistlib.dumps(discover_answer, fmt=plistlib.FMT_BINARY)
+        discover_answer_binary = plistlib.dumps(
+            discover_answer, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
+        )
 
-        AirDropUtil.write_debug(self.config, discover_answer_binary, 'receive_discover_response.plist')
+        AirDropUtil.write_debug(
+            self.config, discover_answer_binary, "receive_discover_response.plist"
+        )
 
         # Change to actual length
         self._set_response(len(discover_answer_binary))
         self.wfile.write(discover_answer_binary)
 
     def handle_ask(self):
-        content_length = int(self.headers['Content-Length'])
+        content_length = int(self.headers["Content-Length"])
         post_data = self.rfile.read(content_length)
 
-        AirDropUtil.write_debug(self.config, post_data, 'receive_ask_request.plist')
+        AirDropUtil.write_debug(self.config, post_data, "receive_ask_request.plist")
 
-        ask_response = {'ReceiverModelName': self.config.computer_model, 'ReceiverComputerName': self.config.computer_name}
-        ask_resp_binary = plistlib.dumps(ask_response, fmt=plistlib.FMT_BINARY)
+        ask_response = {
+            "ReceiverModelName": self.config.computer_model,
+            "ReceiverComputerName": self.config.computer_name,
+        }
+        ask_resp_binary = plistlib.dumps(
+            ask_response, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
+        )
 
-        AirDropUtil.write_debug(self.config, ask_resp_binary, 'receive_ask_response.plist')
+        AirDropUtil.write_debug(
+            self.config, ask_resp_binary, "receive_ask_response.plist"
+        )
 
         self._set_response(len(ask_resp_binary))
         self.wfile.write(ask_resp_binary)
 
     def handle_upload(self):
-        if self.headers.get('content-type', '').lower() != 'application/x-cpio':
-            logger.warning('Unsupported content-type: {}'.format(self.headers.get('content-type')))
+        if self.headers.get("content-type", "").lower() != "application/x-cpio":
+            logger.warning(
+                f"Unsupported content-type: {self.headers.get('content-type')}"
+            )
             self.send_response(406)  # Unprocessable Entity
-            self.send_header('Content-Type', 'application/x-cpio')
-            self.send_header('Content-Length', 0)
-            self.send_header('Connection', 'close')
+            self.send_header("Content-Type", "application/x-cpio")
+            self.send_header("Content-Length", 0)
+            self.send_header("Connection", "close")
             self.end_headers()
             return
 
         # If pipelining is not support, 'Expect: 100-continue' is sent to which we need to respond
-        if self.headers.get('expect', '').lower() == '100-continue':
+        if self.headers.get("expect", "").lower() == "100-continue":
             self.send_response(100)
-            self.send_header('Content-Length', 0)
+            self.send_header("Content-Length", 0)
             self.end_headers()
 
-        if self.headers.get('transfer-encoding', '').lower() != 'chunked':
-            logger.warning('Expect chunked transfer encoding')
+        if self.headers.get("transfer-encoding", "").lower() != "chunked":
+            logger.warning("Expect chunked transfer encoding")
             self.send_response(400)  # Bad Request
-            self.send_header('Transfer-Encoding', 'Chunked')
-            self.send_header('Content-Length', 0)
-            self.send_header('Connection', 'close')
+            self.send_header("Transfer-Encoding", "Chunked")
+            self.send_header("Content-Length", 0)
+            self.send_header("Connection", "close")
             self.end_headers()
             return
 
@@ -269,18 +298,20 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
             with libarchive.read.stream_reader(stream) as archive:
                 libarchive.extract.extract_entries(archive, flags)
 
-        logger.info('Receiving file(s) ...')
+        logger.info("Receiving file(s) ...")
         start = time.time()
         reader = HTTPChunkedReader(self.rfile)
         extract_stream(reader)
 
         transferred = reader.total / 1024.0 / 1024.0
         speed = transferred / (time.time() - start)
-        logger.info('File(s) received (size {:.02f} MB, speed {:.02f} MB/s)'.format(transferred, speed))
+        logger.info(
+            f"File(s) received (size {transferred:.02f} MB, speed {speed:.02f} MB/s)"
+        )
 
         self.send_response(200)
-        self.send_header('Content-Length', 0)
-        self.send_header('Connection', 'close')
+        self.send_header("Content-Length", 0)
+        self.send_header("Connection", "close")
         self.end_headers()
 
     def do_POST(self):
@@ -288,19 +319,23 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
         Handle post requests
         """
 
-        logger.debug('POST request at {}'.format(self.path))
-        logger.debug('Headers\n{}'.format(self.headers))
+        logger.debug(f"POST request at {self.path}")
+        logger.debug(f"Headers\n{self.headers}")
 
-        if self.path == '/Discover':
+        if self.path == "/Discover":
             self.handle_discover()
-        elif self.path == '/Ask':
+        elif self.path == "/Ask":
             self.handle_ask()
-        elif self.path == '/Upload':
+        elif self.path == "/Upload":
             self.handle_upload()
         else:
-            answer = 'POST request for {}'.format(self.path).encode('utf-8')
-            self._set_response(len(answer))
-            self.wfile.write(answer)
+            logger.debug(f"POST request at {self.path}")
+            self.send_response(400)
+            self.send_header("Content-Length", 0)
+            self.end_headers()
 
     def log_message(self, format, *args):
-        logger.debug('{} - - [{}] {}'.format(self.client_address[0], self.log_date_time_string(), format % args))
+        # pylint: disable=redefined-builtin
+        logger.debug(
+            f"{self.client_address[0]} - - [{self.log_date_time_string()}] {format % args}"
+        )

opendrop/util.py

@@ -17,30 +17,25 @@ 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 base64
-import datetime
-import hashlib
 import io
 import ipaddress
 import os
-import plistlib
 
 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 (
+from libarchive.entry import ArchiveEntry, new_archive_entry
+from libarchive.ffi import (  # pylint: disable=no-name-in-module
     ARCHIVE_EOF,
-    entry_sourcepath,
     entry_clear,
-    read_next_header2,
+    entry_sourcepath,
     read_disk_descend,
-    write_header,
+    read_next_header2,
     write_data,
     write_finish_entry,
+    write_get_bytes_per_block,
+    write_header,
 )
 from libarchive.write import ArchiveWrite, new_archive_read_disk
+from PIL import ExifTags, Image
 
 
 class AirDropUtil:
@@ -48,6 +43,7 @@ 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:
         """
@@ -57,98 +53,39 @@ class AirDropUtil:
         """
 
         # Default UTI Type
-        uti_type = 'public.content'
+        uti_type = "public.content"
         if len(flp.mime) == 0 or len(flp.type) == 0:
             return uti_type
 
         mime = flp.mime[0]
         f_type = flp.type[0]
-        if 'image' in mime:
-            uti_type = 'public.image'
+        if "image" in mime:
+            uti_type = "public.image"
 
-            if 'jpg' in mime:
-                uti_type = 'public.jpeg'
-            elif 'jp2' in mime:
-                uti_type = 'public.jpeg-2000'
-            elif 'gif' in mime:
-                uti_type = 'com.compuserve.gif'
-            elif 'png' in mime:
-                uti_type = 'public.png'
-            elif 'raw' in mime or 'raw' in f_type:
-                uti_type = 'public.camera-raw-image'
-        elif 'audio' in f_type:
-            uti_type = 'public.audio'
-        elif 'video' in f_type:
-            uti_type = 'public.video'
-        elif 'archive' in f_type:
-            uti_type = 'public.data'
+            if "jpg" in mime:
+                uti_type = "public.jpeg"
+            elif "jp2" in mime:
+                uti_type = "public.jpeg-2000"
+            elif "gif" in mime:
+                uti_type = "com.compuserve.gif"
+            elif "png" in mime:
+                uti_type = "public.png"
+            elif "raw" in mime or "raw" in f_type:
+                uti_type = "public.camera-raw-image"
+        elif "audio" in f_type:
+            uti_type = "public.audio"
+        elif "video" in f_type:
+            uti_type = "public.video"
+        elif "archive" in f_type:
+            uti_type = "public.data"
 
-            if 'gzip' in mime:
-                uti_type = 'org.gnu.gnu-zip-archive'
-            if 'zip' in mime:
-                uti_type = 'public.zip-archive'
+            if "gzip" in mime:
+                uti_type = "org.gnu.gnu-zip-archive"
+            if "zip" in mime:
+                uti_type = "public.zip-archive"
 
         return uti_type
 
-    @staticmethod
-    def record_data(config, tls_cert, sign_cert, key):
-        """
-        This method generates the sender record data and will sign it using the CMS format.
-
-        This code serves documentation purposes only and is UNTESTED. To be accepted by Apple clients, we would need the
-        Apple-owned private key of the signing certificate.
-
-        :param tls_cert: path to certificate used for AirDrop TLS connections
-        :param sign_cert: path to signing certificate
-        :param key: path to private key to the signing certificate
-        """
-
-        valid_date = datetime.datetime.now() - datetime.timedelta(days=3)
-        valid_date_string = valid_date.strftime('%Y-%m-%dT%H:%M:%SZ')
-
-        emails_hashed = [hashlib.sha256(email.encode('utf-8')).hexdigest() for email in config.email]
-        phone_numbers_hashed = [hashlib.sha256(phone_number.encode('utf-8')).hexdigest() for phone_number in config.phone]
-
-        # Get the common name of the TLS certificate
-        with open(tls_cert, 'rb') as cert_file:
-            cert = x509.X509(cert_file.read())
-            cn = cert.subject[oid.Oid('2.5.4.3')]
-            encDsID = cn.replace('com.apple.idms.appleid.prd.', '')
-
-        # Construct record data
-        record_data = {
-            'Version': 2,
-            'encDsID': encDsID,  # Common name suffix of the certificate
-            'altDsID': encDsID,  # Same as encDsID
-            'SuggestValidDuration': 30 * 24 * 60 * 60,  # in seconds
-            'ValidAsOf': valid_date_string,  # 3 days before now
-            'ValidatedEmailHashes': emails_hashed,
-            'ValidatedPhoneHashes': phone_numbers_hashed,
-        }
-        record_data_plist = plistlib.dumps(record_data, fmt=plistlib.FMT_XML)
-
-        with open(sign_cert, 'rb') as sign_cert_file:
-            with open(key, 'rb') as key_file:
-                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)
-                signed_data = AirDropUtil.pem2der(cms_signed.pem())
-
-        return signed_data
-
-    @staticmethod
-    def pem2der(s):
-        """
-        Create DER Formatted bytes from a PEM Base64 String
-
-        :param s: PEM formatted string
-        """
-        start = s.find('-----\n')
-        finish = s.rfind('\n-----END')
-        data = s[start + 6:finish]
-        return base64.b64decode(data)
-
     @staticmethod
     def generate_file_icon(file_path):
         """
@@ -161,9 +98,13 @@ class AirDropUtil:
 
         # rotate according to EXIF tags
         try:
-            exif = dict((ExifTags.TAGS[k], v) for k, v in im._getexif().items() if k in ExifTags.TAGS)
+            exif = dict(
+                (ExifTags.TAGS[k], v)
+                for k, v in im._getexif().items()  # pylint: disable=protected-access
+                if k in ExifTags.TAGS
+            )
             angles = {3: 180, 6: 270, 8: 90}
-            orientation = exif['Orientation']
+            orientation = exif["Orientation"]
             if orientation in angles.keys():
                 im = im.rotate(angles[orientation], expand=True)
         except (AttributeError, KeyError):
@@ -171,15 +112,15 @@ class AirDropUtil:
 
         # Big image
         im.thumbnail((540, 540), Image.ANTIALIAS)
-        imgByteArr = io.BytesIO()
-        im.save(imgByteArr, format='JPEG2000')
-        file_icon = imgByteArr.getvalue()
+        img_bytes = io.BytesIO()
+        im.save(img_bytes, format="JPEG2000")
+        file_icon = img_bytes.getvalue()
 
         # Small image
         # im.thumbnail((64, 64), Image.ANTIALIAS)
-        # imgByteArr = io.BytesIO()
-        # im.save(imgByteArr, format='JPEG2000')
-        # small_file_icon = imgByteArr.getvalue()
+        # img_bytes = io.BytesIO()
+        # im.save(img_bytes, format='JPEG2000')
+        # small_file_icon = img_bytes.getvalue()
 
         return file_icon
 
@@ -192,6 +133,7 @@ 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:
@@ -204,7 +146,9 @@ class AirDropUtil:
 
         for ip in interface.ips:
             if ip.is_IPv6 and ipv6:
-                return ipaddress.IPv6Address(ip.ip[0])  # first of (ip, flowinfo, scope_id) tuple
+                return ipaddress.IPv6Address(
+                    ip.ip[0]
+                )  # first of (ip, flowinfo, scope_id) tuple
             if ip.is_IPv4 and not ipv6:
                 return ipaddress.IPv4Address(ip.ip)
 
@@ -217,8 +161,8 @@ class AirDropUtil:
         if not os.path.exists(config.debug_dir):
             os.makedirs(config.debug_dir)
         debug_file_path = os.path.join(config.debug_dir, file_name)
-        with open(debug_file_path, 'wb') as file:
-            if hasattr(data, 'read'):
+        with open(debug_file_path, "wb") as file:
+            if hasattr(data, "read"):
                 file.write(data.read())
                 data.seek(0)  # reset cursor position
             else:  # assume bytes-like
@@ -232,7 +176,7 @@ class AbsArchiveWrite(ArchiveWrite):
         """
         write_p = self._pointer
 
-        block_size = ffi.write_get_bytes_per_block(write_p)
+        block_size = write_get_bytes_per_block(write_p)
         if block_size <= 0:
             block_size = 10240  # pragma: no cover
 
@@ -247,7 +191,7 @@ class AbsArchiveWrite(ArchiveWrite):
                     read_disk_descend(read_p)
                     write_header(write_p, entry_p)
                     try:
-                        with open(entry_sourcepath(entry_p), 'rb') as f:
+                        with open(entry_sourcepath(entry_p), "rb") as f:
                             while True:
                                 data = f.read(block_size)
                                 if not data:

requirements-dev.txt

@@ -1,3 +1,6 @@
+black
 flake8
+flake8-bugbear
+isort
+pylint
 pytest
-yapf

setup.cfg

@@ -1,6 +1,17 @@
 [flake8]
-max-line-length = 127
+extend-ignore = E203, E501
+max-line-length = 80
+max-complexity = 18
+select = B9
 
-[yapf]
-based_on_style = pep8
-column_limit = 127
+[isort]
+multi_line_output = 3
+include_trailing_comma = True
+force_grid_wrap = 0
+use_parentheses = True
+ensure_newline_before_comments = True
+line_length = 88
+
+[pylint]
+disable = C, R, W0511, W1203
+max-line-length = 88

setup.py

@@ -4,41 +4,49 @@ from os.path import abspath, dirname, join
 from setuptools import find_packages, setup
 
 this_dir = abspath(dirname(__file__))
-with open(join(this_dir, 'README.md'), encoding='utf-8') as file:
+with open(join(this_dir, "README.md"), encoding="utf-8") as file:
     long_description = file.read()
 
 setup(
-    name='opendrop',
+    name="opendrop",
     version=__version__,
-    python_requires='>=3.6',
-    description='An open Apple AirDrop implementation',
+    python_requires=">=3.6",
+    description="An open Apple AirDrop implementation",
     long_description=long_description,
-    long_description_content_type='text/markdown',
-    url='https://owlink.org',
+    long_description_content_type="text/markdown",
+    url="https://owlink.org",
     project_urls={
-        'Source': 'https://github.com/seemoo-lab/opendrop',
-        'Research Paper': 'https://usenix.org/conference/usenixsecurity19/presentation/stute',
+        "Source": "https://github.com/seemoo-lab/opendrop",
+        "Research Paper": "https://usenix.org/conference/usenixsecurity19/presentation/stute",
     },
-    author='The Open Wireless Link Project',
-    author_email='mstute@seemoo.tu-darmstadt.de',
+    author="The Open Wireless Link Project",
+    author_email="mstute@seemoo.tu-darmstadt.de",
     classifiers=[
-        'License :: OSI Approved :: GNU General Public License v3 (GPLv3)',
-        'Operating System :: MacOS',
-        'Operating System :: POSIX :: Linux',
-        'Natural Language :: English',
-        'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.6',
-        'Programming Language :: Python :: 3.7',
+        "License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
+        "Operating System :: MacOS",
+        "Operating System :: POSIX :: Linux",
+        "Natural Language :: English",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
     ],
-    keywords='cli',
-    packages=find_packages(exclude=['docs']),
-    package_data={'opendrop': ['certs/*.pem']},
+    keywords="cli",
+    packages=find_packages(exclude=["docs"]),
+    package_data={"opendrop": ["certs/*.pem"]},
     install_requires=[
-        'Pillow', 'ctypescrypto', 'fleep', 'ifaddr', 'libarchive-c', 'requests', 'requests_toolbelt', 'zeroconf>=0.24.2'
+        "Pillow",
+        "fleep",
+        "ifaddr",
+        "libarchive-c",
+        "requests",
+        "requests_toolbelt",
+        "zeroconf>=0.24.2",
     ],
     entry_points={
-        'console_scripts': [
-            'opendrop=opendrop.cli:main',
+        "console_scripts": [
+            "opendrop=opendrop.cli:main",
         ],
     },
 )

tests/test_client.py

@@ -4,15 +4,16 @@ from opendrop.config import AirDropConfig
 
 def get_loopback():
     import ifaddr
+
     for adapter in ifaddr.get_adapters():
-        if adapter.name.startswith('lo'):
+        if adapter.name.startswith("lo"):
             return adapter.name
     return None
 
 
 def test_browser_setup():
     loopback = get_loopback()
-    assert loopback is not None, 'Could not find loopback interface'
+    assert loopback is not None, "Could not find loopback interface"
     config = AirDropConfig(interface=loopback)
     browser = AirDropBrowser(config)
     browser.start()

tests/test_server.py

@@ -4,15 +4,16 @@ from opendrop.config import AirDropConfig
 
 def get_loopback():
     import ifaddr
+
     for adapter in ifaddr.get_adapters():
-        if adapter.name.startswith('lo'):
+        if adapter.name.startswith("lo"):
             return adapter.name
     return None
 
 
 def test_server_setup():
     loopback = get_loopback()
-    assert loopback is not None, 'Could not find loopback interface'
+    assert loopback is not None, "Could not find loopback interface"
     config = AirDropConfig(interface=loopback)
     server = AirDropServer(config)
     server.start_service()