BRynRrKw

1from pathlib import Path
2from datetime import datetime, timezone
3import requests
4import dataclasses
5import json
6import subprocess
7import shutil
8import sys
9
10api_version = "v0a769"
11api = f"https://api.cloudflareclient.com/{api_version}"
12reg_url = f"{api}/reg"
13status_url = f"{api}/client_config"
14terms_of_service_url = "https://www.cloudflare.com/application/terms/"
15
16data_path = Path(".")
17identity_path = data_path.joinpath("wgcf-identity.json")
18config_path = data_path.joinpath("wgcf-profile.conf")
19
20default_headers = {"Accept-Encoding": "gzip",
21                   "User-Agent": "okhttp/3.12.1"}
22
23# toggle to allow sniffing traffic
24debug = False
25
26
27def get_verify() -> bool:
28    return not debug
29
30
31def get_config_url(account_token: str) -> str:
32    return f"{reg_url}/{account_token}"
33
34
35@dataclasses.dataclass
36class AccountData():
37    account_id: str
38    access_token: str
39    private_key: str
40
41
42@dataclasses.dataclass
43class ConfigurationData():
44    local_address_ipv4: str
45    local_address_ipv6: str
46    endpoint_address_host: str
47    endpoint_address_ipv4: str
48    endpoint_address_ipv6: str
49    endpoint_public_key: str
50    warp_enabled: bool
51    account_type: str
52    warp_plus_enabled: bool
53
54
55def get_timestamp() -> str:
56    # SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss", Locale.US)
57    timestamp = datetime.now(tz=timezone.utc).astimezone(None).strftime("%Y-%m-%dT%H:%M:%S.%f%z")
58    # trim microseconds to 2 digits
59    timestamp = timestamp[:-10]+timestamp[-6:]
60    # separate timezone offset
61    timestamp = timestamp[:-2]+":"+timestamp[-2:]
62    return timestamp
63
64
65def gen_private_key() -> str:
66    result: str = subprocess.run(["wg", "genkey"], capture_output=True).stdout.decode('utf-8')
67    return result.strip()
68
69
70def gen_public_key(private_key: str) -> str:
71    result: str = subprocess.run(["wg", "pubkey"], input=bytes(private_key, 'utf-8'),
72                                 capture_output=True).stdout.decode('utf-8')
73    return result.strip()
74
75
76def do_register() -> AccountData:
77    timestamp = get_timestamp()
78    private_key = gen_private_key()
79    public_key = gen_public_key(private_key)
80    data = {"install_id": "", "tos": timestamp, "key": public_key, "fcm_token": "", "type": "Android",
81            "locale": "en_US"}
82
83    headers = default_headers.copy()
84    headers["Content-Type"] = "application/json; charset=UTF-8"
85
86    response = requests.post(reg_url, json=data, headers=headers, verify=get_verify())
87
88    response.raise_for_status()
89    response = json.loads(response.content)
90    return AccountData(response["id"], response["token"], private_key)
91
92
93def save_identitiy(account_data: AccountData):
94    with open(identity_path, "w") as f:
95        f.write(json.dumps(dataclasses.asdict(account_data), indent=4))
96
97
98def load_identity() -> AccountData:
99    with open(identity_path, "r") as f:
100        account_data = AccountData(**json.loads(f.read()))
101        return account_data
102
103
104def enable_warp(account_data: AccountData):
105    data = {"warp_enabled": True}
106
107    headers = default_headers.copy()
108    headers["Authorization"] = f"Bearer {account_data.access_token}"
109    headers["Content-Type"] = "application/json; charset=UTF-8"
110
111    response = requests.patch(get_config_url(account_data.account_id), json=data, headers=headers, verify=get_verify())
112
113    response.raise_for_status()
114    response = json.loads(response.content)
115    assert response["warp_enabled"] == True
116
117
118def get_server_conf(account_data: AccountData) -> ConfigurationData:
119    headers = default_headers.copy()
120    headers["Authorization"] = f"Bearer {account_data.access_token}"
121
122    response = requests.get(get_config_url(account_data.account_id), headers=headers, verify=get_verify())
123
124    response.raise_for_status()
125    response = json.loads(response.content)
126
127    addresses = response["config"]["interface"]["addresses"]
128    peer = response["config"]["peers"][0]
129    endpoint = peer["endpoint"]
130
131    account = response["account"] if "account" in response else ""
132    account_type = account["account_type"] if account != "" else "free"
133    warp_plus = account["warp_plus"] if account != "" else False
134
135    return ConfigurationData(addresses["v4"], addresses["v6"], endpoint["host"], endpoint["v4"],
136                             endpoint["v6"], peer["public_key"], response["warp_enabled"], account_type, warp_plus)
137
138
139def get_wireguard_conf(private_key: str, address_1: str, address_2: str, public_key: str, endpoint: str) -> str:
140    return f"""
141[Interface]
142PrivateKey = {private_key}
143DNS = 1.1.1.1
144DNS = 1.0.0.1
145Address = {address_1}
146Address = {address_2}
147
148[Peer]
149PublicKey = {public_key}
150AllowedIPs = 0.0.0.0/0
151AllowedIPs = ::/0
152Endpoint = {endpoint}
153"""[1:-1]
154
155
156def create_conf(account_data: AccountData, conf_data: ConfigurationData):
157    with open(config_path, "w") as f:
158        f.write(
159            get_wireguard_conf(account_data.private_key, conf_data.local_address_ipv4,
160                               conf_data.local_address_ipv6, conf_data.endpoint_public_key,
161                               conf_data.endpoint_address_host))
162
163
164if __name__ == "__main__":
165    if shutil.which("wg") == None:
166        print("Error: 'wg' must be installed and added to PATH")
167        print("More information: https://www.wireguard.com/install/")
168        sys.exit(1)
169
170    data_path.mkdir(exist_ok=True)
171    account_data: AccountData
172
173
174    if not identity_path.exists():
175        print("This project is in no way affiliated with Cloudflare!")
176        print(f"Cloudflare's Terms of Service: {terms_of_service_url}")
177        #if not input("Do you agree? (y/N): ").lower() == "y":
178            #sys.exit(2)
179
180        print(f"Creating new identity...")
181        account_data = do_register()
182        save_identitiy(account_data)
183    else:
184        print(f"Loading existing identity...")
185        account_data = load_identity()
186
187    print(f"Getting configuration...")
188    conf_data = get_server_conf(account_data)
189
190    if not conf_data.warp_enabled:
191        print(f"Enabling Warp...")
192        enable_warp(account_data)
193        conf_data.warp_enabled = True
194
195    print(f"Account type: {conf_data.account_type}")
196    print(f"Warp+ enabled: {conf_data.warp_plus_enabled}")
197
198    print("Creating WireGuard configuration...")
199    create_conf(account_data, conf_data)