QQ-Tunnel/data_cap.py at main · patterniha/QQ-Tunnel · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
import sys

from utility.base32 import b32encode_nopad_lower, BASE32_LIST_LOWER, number_to_base32_lower, base32_to_number, \
    BASE32_LOOKUP
from utility.dns import insert_dots


def compute_max_m(s: int, max_allowed: int) -> int:
    """
    Find maximum m such that: m + ⌈m / s⌉ ≤ max_allowed
    """
    if max_allowed <= 0:
        return 0

    q = max_allowed // (s + 1)
    remaining = max_allowed - q * (s + 1)
    r = max(0, remaining - 1)

    return q * s + r


def get_chunk_len(max_encoded_domain_len: int, qname_encoded_len: int, max_sub_len: int, data_offset_width=4) -> int:
    max_allowed = max_encoded_domain_len - qname_encoded_len
    m = compute_max_m(max_sub_len, max_allowed)
    chunk_len = m - data_offset_width - 2  # fragment_part_width is 2
    if chunk_len <= 0:
        raise ValueError("max_encoded_domain_len is too small to fit any data")
    return chunk_len


def get_base32_final_domains(data: bytes, data_offset: int, send_domain_with_chunk_idx: int,
                             send_domain_with_chunk_list: list, max_sub_len: int,
                             data_offset_width: int,
                             max_encoded_domain_len: int) -> \
        list[bytes]:
    data = b32encode_nopad_lower(data)
    final_b_domains = []
    i = 0
    c_loop = True
    s_index = 0
    len_data = len(data)
    data_offset_bytes = number_to_base32_lower(data_offset, data_offset_width)
    while c_loop:
        if i == 64:
            print("ERROR: max_domain_len is too small, packet is not sent, len:", len(data))
            return []
        qname_encoded, chunk_len = send_domain_with_chunk_list[send_domain_with_chunk_idx]
        send_domain_with_chunk_idx = (send_domain_with_chunk_idx + 1) % len(send_domain_with_chunk_list)
        chunk_data = data[s_index:s_index + chunk_len]
        s_index += chunk_len
        if s_index < len_data:
            if i & 32:
                chunk_data = b"".join((data_offset_bytes, BASE32_LIST_LOWER[i & 31], b"8", chunk_data))
            else:
                chunk_data = b"".join((data_offset_bytes, BASE32_LIST_LOWER[i], b"0", chunk_data))
        else:
            if i & 32:
                chunk_data = b"".join((data_offset_bytes, BASE32_LIST_LOWER[i & 31], b"9", chunk_data))
            else:
                chunk_data = b"".join((data_offset_bytes, BASE32_LIST_LOWER[i], b"1", chunk_data))
            c_loop = False
        final_domain = insert_dots(chunk_data, max_sub_len) + qname_encoded
        if len(final_domain) > max_encoded_domain_len:
            sys.exit("Calculation Error!!!")
        final_b_domains.append(final_domain)
        i += 1

    return final_b_domains


def get_chunk_data(data: bytes, data_offset_width: int):
    data_offset = base32_to_number(data[:data_offset_width])

    fragment_part_raw = BASE32_LOOKUP[data[data_offset_width]]
    if fragment_part_raw < 0:
        raise ValueError("Invalid base32 character in fragment part")

    magic = data[data_offset_width + 1]
    if magic == 48:  # b"0"
        fragment_part = fragment_part_raw
        last_fragment = False
    elif magic == 49:  # b"1"
        fragment_part = fragment_part_raw
        last_fragment = True
    elif magic == 56:  # b"8"
        fragment_part = fragment_part_raw | 32
        last_fragment = False
    elif magic == 57:  # b"9"
        fragment_part = fragment_part_raw | 32
        last_fragment = True
    else:
        raise ValueError("Unknown magic")

    e_data = data[data_offset_width + 2:]
    return data_offset, fragment_part, last_fragment, e_data