bigsnitch/mitmaddon/test_bigsnitch.py

#!/usr/bin/env python3

import pdb

import pytest
from networkthread import bPacket, bRequest, bResponse, bHeader, NetworkThread
import os
import tempfile
from queue import Queue
import zmq

from deepdiff import DeepDiff

# usual flow state of the request with some big parts removed
@pytest.fixture
def flowstate_request():
    return {'client_conn': {'address': ('::ffff:127.0.0.1', 60630, 0, 0),
                 'alpn_proto_negotiated': b'http/1.1',
                 'cipher_name': 'TLS_AES_256_GCM_SHA384',
                 'clientcert': None,
                 'id': '5dde7ef8-9b1a-4b60-9d15-d308442a27ea',
                 'mitmcert': '',
                 'sni': 'yolo.jetzt',
                 'timestamp_end': None,
                 'timestamp_start': 1619390481.8003347,
                 'timestamp_tls_setup': 1619390482.6879823,
                 'tls_established': True,
                 'tls_extensions': [],
                 'tls_version': 'TLSv1.3'},
 'error': None,
 'id': '51215b69-c76f-4ac2-afcb-da3b823d9f88',
 'intercepted': False,
 'is_replay': None,
 'marked': False,
 'metadata': {},
 'mode': 'transparent',
 'request': {'authority': b'',
             'content': b'',
             'headers': ((b'Host', b'yolo.jetzt'),
                         (b'User-Agent', b'curl/7.75.0'),
                         (b'Accept', b'*/*')),
             'host': 'yolo.jetzt',
             'http_version': b'HTTP/1.1',
             'method': b'GET',
             'path': b'/',
             'port': 443,
             'scheme': b'https',
             'timestamp_end': 1619390482.69,
             'timestamp_start': 1619390482.6886377,
             'trailers': None},
 'response': None,
 'server_conn': {'address': ('yolo.jetzt', 443),
                 'alpn_proto_negotiated': b'http/1.1',
                 'cert': '',
                 'id': 'ecc4cd3b-7e35-4815-b618-5931fe64729b',
                 'ip_address': ('95.156.226.69', 443),
                 'sni': 'yolo.jetzt',
                 'source_address': ('192.168.42.182', 51514),
                 'timestamp_end': None,
                 'timestamp_start': 1619390481.8154442,
                 'timestamp_tcp_setup': 1619390481.994565,
                 'timestamp_tls_setup': 1619390482.6819758,
                 'tls_established': True,
                 'tls_version': 'TLSv1.2',
                 'via': None},
 'type': 'http',
 'version': 9}

class MitmAddonTestServer:
    def __init__(self, queue, path: str):
        self.queue = queue
        self.path = path
        self.socket = None
        self.context = zmq.Context()
        self.connect()

    def connect(self):
        self.socket = self.context.socket(zmq.PAIR)
        self.socket.connect(self.path)

    def disconnect(self):
        self.socket.setsockopt(zmq.LINGER,0)
        self.socket.close()

    def send_packet(self, pkg: bPacket):
        msg = {"type": pkg.ptype, "id": pkg.flowid, "data": pkg.data}
        self.send(msg)

@pytest.fixture
def client_server():
    queue = Queue()
    sock = os.path.join(tempfile.mkdtemp(), "bigsnitchtest.sock")
    sock = f"ipc://{sock}"
    client = NetworkThread(name="testthread",queue=queue,path=sock)
    client.daemon = True
    client.start()
    server = MitmAddonTestServer(queue, sock)
    server.connect()
    yield client, server
    client.join(1)
    server.disconnect()

class TestBigSnitchWrapper:
    def test_request_convert(self, flowstate_request):
        req = bRequest(flow=flowstate_request)
        d = {'content': b'',
         'headers': [bHeader(key=b'Host', value=b'yolo.jetzt'),
                     bHeader(key=b'User-Agent', value=b'curl/7.75.0'),
                     bHeader(key=b'Accept', value=b'*/*')],
         'host': 'yolo.jetzt',
         'http_version': b'HTTP/1.1',
         'method': b'GET',
         'port': 443,
         'scheme': b'https',
         'server_ip_address': '95.156.226.69',
         'timestamp_end': 1619390482.69,
         'timestamp_start': 1619390482.6886377,
         'tls': True}
        assert not DeepDiff(req.__dict__, d)
"""
class TestMitmAddon:
    def test_request(self, client_server):
        self.client, self.server = client_server
        # create request
        flowitem = FlowItem(bFlowState.UNSENT_HTTP_REQUEST, flow)
        self.q.put_nowait((flow.id, flowitem))
"""