TySS-Dev/opendrop
1
0
Fork 0
mirror of https://github.com/seemoo-lab/opendrop.git synced 2026-09-14 15:16:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

Replace yapf with black as formatter

Browse Source
This commit is contained in:
Milan Stute
2020-11-30 19:52:15 +01:00
parent 5e2ff4210f
commit c926eaf710
13 changed files with 480 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/testpackage.yml
+3 -3
View File
@@ -22,10 +22,10 @@ jobs:
- name: Install package
run: |
pip install -e .
- name: Check format with yapf
- name: Check format with black
run: |
pip install yapf
yapf . -r --diff
pip install black
black . --check --diff
- name: Lint with flake8
run: |
pip install flake8
Makefile
+2 -2
View File
@@ -18,7 +18,7 @@ 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)
@@ -27,4 +27,4 @@ test: $(VENV)
$(PYTHON) -m pytest
autoformat: $(VENV)
$(PYTHON) -m yapf . -r --in-place --exclude $(VENV)
$(PYTHON) -m black . --exclude $(VENV)
opendrop/__init__.py
+8 -6
View File
@@ -21,12 +21,14 @@ import logging
import os
import platform
__version__ = '0.11.0'
__version__ = "0.11.0"
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
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
+81 -47
View File
@@ -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,
self.config = AirDropConfig(
email=args.email,
phone=args.phone,
computer_name=args.name,
computer_model=args.model,
debug=args.debug,
interface=args.interface)
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,7 +113,7 @@ class AirDropCli:
self.server.stop()
def find(self):
logger.info('Looking for receivers. Press enter to stop ...')
logger.info("Looking for receivers. Press enter to stop ...")
self.browser = AirDropBrowser(self.config)
self.browser.start(callback_add=self._found_receiver)
try:
@@ -101,27 +122,33 @@ class AirDropCli:
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(
"Save discovery results to {}".format(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.warn("Ignoring receiver with missing address {}".format(info))
return
id = info.name.split('.')[0]
id = info.name.split(".")[0]
hostname = info.server
port = int(info.port)
logger.debug('AirDrop service found: {}, {}:{}, ID {}'.format(hostname, address, port, id))
logger.debug(
"AirDrop service found: {}, {}:{}, ID {}".format(
hostname, address, port, 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 +164,21 @@ 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": id,
"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(
"Found index {} ID {} name {}".format(index, id, receiver_name)
)
else:
logger.debug('Receiver ID {} is not discoverable'.format(id))
logger.debug("Receiver ID {} is not discoverable".format(id))
self.lock.release()
def receive(self):
@@ -161,26 +190,29 @@ 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(
"Old discovery report (%.1f seconds), consider running 'opendrop find' again",
age,
)
with open(self.config.discovery_report, "r") as f:
infos = json.load(f)
# (1) try 'index'
@@ -194,12 +226,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
+84 -57
View File
@@ -39,15 +39,23 @@ 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(
"Interface {} does not have an IPv6 address. "
"Make sure that `owl` is running.".format(config.interface)
)
else:
raise RuntimeError('Interface {} does not have an IPv6 address'.format(config.interface))
raise RuntimeError(
"Interface {} does not have an IPv6 address".format(
config.interface
)
)
self.zeroconf = Zeroconf(interfaces=[str(self.ip_addr)],
self.zeroconf = Zeroconf(
interfaces=[str(self.ip_addr)],
ip_version=IPVersion.V6Only,
apple_p2p=platform.system() == 'Darwin')
apple_p2p=platform.system() == "Darwin",
)
self.callback_add = None
self.callback_remove = None
@@ -61,7 +69,7 @@ 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()
@@ -70,13 +78,13 @@ class AirDropBrowser:
def add_service(self, zeroconf, type, name):
info = zeroconf.get_service_info(type, name)
logger.debug('Add service {}'.format(name))
logger.debug("Add service {}".format(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))
logger.debug("Remove service {}".format(name))
if self.callback_remove is not None:
self.callback_remove(info)
@@ -89,9 +97,11 @@ class AirDropClient:
self.http_conn = None
def send_POST(self, url, body, headers=None):
logger.debug('Send {} request'.format(url))
logger.debug("Send {} request".format(url))
AirDropUtil.write_debug(self.config, body, 'send_{}_request.plist'.format(url.lower().strip('/')))
AirDropUtil.write_debug(
self.config, body, "send_{}_request.plist".format(url.lower().strip("/"))
)
_headers = self._get_headers()
if headers is not None:
@@ -99,76 +109,82 @@ class AirDropClient:
_headers[key] = val
if self.http_conn is None:
# Use single connection
self.http_conn = HTTPSConnectionAWDL(self.receiver_host,
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)
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,
"send_{}_response.plist".format(url.lower().strip("/")),
)
if http_resp.status != 200:
status = False
logger.debug('{} request failed: {}'.format(url, http_resp.status))
logger.debug("{} request failed: {}".format(url, http_resp.status))
else:
status = True
logger.debug('{} request successful'.format(url))
logger.debug("{} request successful".format(url))
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)
success, 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)
success, _ = self.send_POST("/Ask", ask_binary)
return success
@@ -177,20 +193,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 +221,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,7 +235,9 @@ class HTTPSConnectionAWDL(HTTPSConnection):
"""
This class allows to bind the HTTPConnection to a specific network interface
"""
def __init__(self,
def __init__(
self,
host,
port=None,
key_file=None,
@@ -225,29 +247,34 @@ class HTTPSConnectionAWDL(HTTPSConnection):
*,
context=None,
check_hostname=None,
interface_name=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,
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)
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,
@@ -269,7 +296,7 @@ class HTTPSConnectionAWDL(HTTPSConnection):
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 +312,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
+53 -23
View File
@@ -36,6 +36,7 @@ class AirDropReceiverFlags:
according to sharingd`[SDBonjourBrowser removeInvalidNodes:].
Default flags on macOS: 0x3fb according to sharingd`[SDRapportBrowser defaultSFNodeFlags]
"""
SUPPORTS_URL = 0x01
SUPPORTS_DVZIP = 0x02
SUPPORTS_PIPELINING = 0x04
@@ -43,26 +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,
def __init__(
self,
host_name=None,
computer_name=None,
computer_model=None,
server_port=8771,
airdrop_dir='~/.opendrop',
airdrop_dir="~/.opendrop",
service_id=None,
email=None,
phone=None,
debug=False,
interface=None):
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()
@@ -71,19 +76,21 @@ 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 = "{0:0{1}x}".format(
random.randint(0, 0xFFFFFFFFFFFF), 12
) # 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:
@@ -94,39 +101,62 @@ 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(
"Need Apple root CA certificate: {}".format(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_data = None
def create_default_key(self):
logger.info('Create new self-signed certificate in {}'.format(self.key_dir))
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)
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)
stderr=subprocess.PIPE,
)
def get_ssl_context(self):
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS) # lgtm[py/insecure-protocol], TODO see https://github.com/Semmle/ql/issues/2554
ctx = ssl.SSLContext(
ssl.PROTOCOL_TLS
) # lgtm[py/insecure-protocol], TODO see https://github.com/Semmle/ql/issues/2554
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
+101 -61
View File
@@ -39,47 +39,63 @@ 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(
"Interface {} does not have an IPv6 address. "
"Make sure that `owl` is running.".format(self.config.interface)
)
else:
raise RuntimeError('Interface {} does not have an IPv6 address'.format(self.config.interface))
raise RuntimeError(
"Interface {} does not have an IPv6 address".format(
self.config.interface
)
)
self.Handler = AirDropServerHandler
self.Handler.config = self.config
self.zeroconf = Zeroconf(interfaces=[str(self.ip_addr)],
self.zeroconf = Zeroconf(
interfaces=[str(self.ip_addr)],
ip_version=IPVersion.V6Only,
apple_p2p=platform.system() == 'Darwin')
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.',
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])
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(
"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):
@@ -92,15 +108,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 +126,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 +138,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 +147,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 +155,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("GET request at {}".format(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 +205,70 @@ 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
)
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_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')
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(
"Unsupported content-type: {}".format(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 +301,22 @@ 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(
"File(s) received (size {:.02f} MB, speed {:.02f} MB/s)".format(
transferred, speed
)
)
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,20 +324,24 @@ class AirDropServerHandler(BaseHTTPRequestHandler):
Handle post requests
"""
logger.debug('POST request at {}'.format(self.path))
logger.debug('Headers\n{}'.format(self.headers))
logger.debug("POST request at {}".format(self.path))
logger.debug("Headers\n{}".format(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:
logger.debug('POST request at {}'.format(self.path))
logger.debug("POST request at {}".format(self.path))
self.send_response(400)
self.send_header('Content-Length', 0)
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))
logger.debug(
"{} - - [{}] {}".format(
self.client_address[0], self.log_date_time_string(), format % args
)
)
opendrop/util.py
+68 -49
View File
@@ -48,6 +48,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,36 +58,36 @@ 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
@@ -104,35 +105,46 @@ class AirDropUtil:
"""
valid_date = datetime.datetime.now() - datetime.timedelta(days=3)
valid_date_string = valid_date.strftime('%Y-%m-%dT%H:%M:%SZ')
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]
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:
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.', '')
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,
"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:
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)
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
@@ -144,9 +156,9 @@ class AirDropUtil:
:param s: PEM formatted string
"""
start = s.find('-----\n')
finish = s.rfind('\n-----END')
data = s[start + 6:finish]
start = s.find("-----\n")
finish = s.rfind("\n-----END")
data = s[start + 6 : finish]
return base64.b64decode(data)
@staticmethod
@@ -161,9 +173,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()
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):
@@ -172,7 +188,7 @@ class AirDropUtil:
# Big image
im.thumbnail((540, 540), Image.ANTIALIAS)
imgByteArr = io.BytesIO()
im.save(imgByteArr, format='JPEG2000')
im.save(imgByteArr, format="JPEG2000")
file_icon = imgByteArr.getvalue()
# Small image
@@ -192,6 +208,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 +221,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 +236,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
@@ -247,7 +266,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
+2 -1
View File
@@ -1,3 +1,4 @@
black
flake8
flake8-bugbear
pytest
yapf
setup.cfg
+4 -5
View File
@@ -1,6 +1,5 @@
[flake8]
max-line-length = 127
[yapf]
based_on_style = pep8
column_limit = 127
extend-ignore = E203, E501
max-line-length = 80
max-complexity = 18
select = B9
setup.py
+30 -23
View File
@@ -4,41 +4,48 @@ 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",
],
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",
"ctypescrypto",
"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
+3 -2
View File
@@ -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
+3 -2
View File
@@ -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()