Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Select an option

  • Save hungtoong/8a544d1e375d4ed86dcb13bee4b3b560 to your computer and use it in GitHub Desktop.

Select an option

Save hungtoong/8a544d1e375d4ed86dcb13bee4b3b560 to your computer and use it in GitHub Desktop.
Generate wireguard config from cloudflare zero trust
"""
Generate wireguard config from cloudflare zero trust
Credit to https://gitlab.com/Misaka-blog/warp-script
"""
import datetime
import http.cookiejar
import json
import random
import re
import string
from urllib import parse, request
"""
Parameters
TEAM_NAME, PRIVATE_KEY and PUBLIC_KEY need to be set
Note: this script will hang up and ask you for the CODE sent to your EMAIL
FAQ:
If "HTTP 401": TOKEN has expired, need a new one
If "HTTP 409": PRIVATE_KEY and PUBLIC_KEY have already been used, need a new pair
"""
# Your Zero Trust team name
TEAM_NAME = ""
# Email address to receive the CODE
EMAIL_ADDR = ""
# Generate public key with `wg genkey`
PRIVATE_KEY = ""
# Generate public key with `echo <private key> | wg pubkey`
PUBLIC_KEY = ""
# Optional, if you get the TOKEN manually
TOKEN = ""
# Output filename
OUTPUT_FILENAME = "cf_zero_trust.conf"
"""
Fetch TOKEN
"""
token = ""
if not TOKEN:
print("Initialising...")
opener = request.build_opener(
request.HTTPCookieProcessor(http.cookiejar.CookieJar())
) # Cookies needed for auth
# Get the first cookie
url = f"https://{TEAM_NAME}.cloudflareaccess.com/warp"
req = request.Request(url)
print("Fetching cookies...")
with opener.open(req) as resp: # Get register end point
# Get the second cookie
url = resp.url.replace("/login/", "/verify-code/")
req = request.Request(
url,
data=parse.urlencode(
{
"email": EMAIL_ADDR,
"client_id": "",
"connector_id": "",
"connector_type": "",
"redirect_url": "",
}
).encode("ascii"),
headers={"Content-Type": "application/x-www-form-urlencoded"},
method="POST",
)
print("Registering...")
with opener.open(req) as resp: # Send email address
code = input("Please enter the CODE: ") # Ask for the code
# Get the final result
qs = parse.parse_qs(parse.urlparse(resp.url).query)
url = f"https://{TEAM_NAME}.cloudflareaccess.com/cdn-cgi/access/callback"
req = request.Request(
url,
data=parse.urlencode({"code": code, "nonce": qs["nonce"][0]}).encode(
"ascii"
),
headers={"Content-Type": "application/x-www-form-urlencoded"},
method="POST",
)
with opener.open(req) as resp: # Send code
html = resp.read(4096).decode("utf-8") # 4096 is long enough to get the TOKEN
token = re.findall('<meta http-equiv="refresh" content=.*token=(.*)" />', html)[
0
]
"""
Generation
"""
print("Initialising...")
install_id = "".join(random.choices(string.ascii_letters + string.digits, k=22))
fcm_token = f'{install_id}:APA91b{"".join(random.choices(string.ascii_letters + string.digits, k=134))}'
headers = {
"User-Agent": "okhttp/3.12.1",
"CF-Client-Version": "a-6.10-2158",
"Content-Type": "application/json",
"Cf-Access-Jwt-Assertion": TOKEN if TOKEN else token,
}
data = {
"key": PUBLIC_KEY,
"install_id": install_id,
"fcm_token": fcm_token,
"tos": datetime.datetime.now().isoformat()[:-3] + "Z",
"model": "Linux",
"name": install_id, # Or you name it
"serial_number": install_id, # Or you name it
"locale": "zh_CN",
}
req = request.Request(
"https://api.cloudflareclient.com/v0a2158/reg",
data=json.dumps(data).encode(),
headers=headers,
) # Register with API version v0a2158
print("Generating...")
with request.urlopen(req) as resp:
v6_addr = json.load(resp)["config"]["interface"]["addresses"]["v6"]
with open(OUTPUT_FILENAME, "w", newline="") as f:
f.writelines(
[
"[Interface]",
f"PrivateKey = {PRIVATE_KEY}",
f"Address = 172.16.0.2/32, {v6_addr}/128",
"DNS = 1.1.1.1, 1.0.0.1, 2606:4700:4700::1111, 2606:4700:4700::1001",
"MTU = 1280",
"",
"[Peer]",
"PublicKey = bmXOC+F1FxEMF9dyiK2H5/1SUtzH0JuVo51h2wPfgyo=",
"AllowedIPs = 0.0.0.0/0, ::/0",
"Endpoint = engage.cloudflareclient.com:2408",
]
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment