首先第一步骤,注册一个KataBump账号(应该很多人都知道,就是能获取法国二级域名那个平台),账号还是比较好注册的,随便一个邮箱就可以,注册链接:https://dashboard.katabump.com/auth/login第二步登入KataBump平台,点击Ordera按钮领取免费的PY容器,Name随便填,Type勾选Python,Resources默认就行(第一个才是免费的,其余的也创建不了)然后过人机验证,点确认.
第三步,机器创建完,首页有个 Your servers,点Action下面的红色see就能找到你的机器信息,然后获取你的翼龙面板,点击有图标Access server就能找到翼龙面板的登入网址和账号密码。
第四步.登入翼龙面板。点击Files进入文件,点击的NEW Files添加启动文件(没有启动文件无法启动容器)启动文件类型选Python,文件名app.py.(再添加一个requirements.txt的空文件)就可以按绿色按钮启动容器了。
http版本启动文件:
import socket
import threading
import select
import base64
import os
PORT = int(os.environ.get('SERVER_PORT', os.environ.get('PORT', 20006)))# ================= 账号密码配置 =================AUTH_USER = "admin"AUTH_PASS = "123456"# ===============================================def handle_client(client_sock, client_addr): print(f"[+] 收到客户端连接: {client_addr}") remote_sock = None try: # 读取 HTTP 请求头 request_data = b"" while b"\r\n\r\n" not in request_data: chunk = client_sock.recv(4096) if not chunk: break request_data += chunk if len(request_data) > 8192: break if not request_data: client_sock.close() return lines = request_data.split(b"\r\n") first_line = lines[0].decode('utf-8', errors='ignore') parts = first_line.split(" ") if len(parts) < 2: client_sock.close() return method, url = parts[0], parts[1] # 检查 Proxy-Authorization 认证 authed = False expected_auth = base64.b64encode(f"{AUTH_USER}:{AUTH_PASS}".encode()).decode() for line in lines: if line.lower().startswith(b"proxy-authorization: basic "): token = line.split(b" ")[2].decode() if token == expected_auth: authed = True break if not authed: # 认证失败,返回 407 要求输入账号密码 challenge = ( "HTTP/1.1 407 Proxy Authentication Required\r\n" "Proxy-Authenticate: Basic realm=\"Proxy Required\"\r\n" "Content-Length: 0\r\n\r\n" ) client_sock.sendall(challenge.encode()) client_sock.close() return # 解析目标地址 if method == "CONNECT": # HTTPS / 隧道代理 host_port = url if ":" in host_port: target_host, target_port = host_port.split(":") target_port = int(target_port) else: target_host = host_port target_port = 443 # 回复客户端 CONNECT 建立成功 client_sock.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n") else: # 普通 HTTP 代理 if url.startswith("http://"): url = url[7:] if "/" in url: host_port, _ = url.split("/", 1) else: host_port = url if ":" in host_port: target_host, target_port = host_port.split(":") target_port = int(target_port) else: target_host = host_port target_port = 80 print(f"[->] 正在转发到目标: {target_host}:{target_port}") remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) remote_sock.connect((target_host, target_port)) if method != "CONNECT": new_first_line = f"{method} {url} HTTP/1.1\r\n" forward_data = new_first_line.encode() + b"\r\n".join([l for l in lines[1:] if not l.lower().startswith(b"proxy-")]) remote_sock.sendall(forward_data) sockets = [client_sock, remote_sock] while True: r, w, e = select.select(sockets, [], [], 60) if not r: break if client_sock in r: data = client_sock.recv(4096) if not data: break remote_sock.sendall(data) elif remote_sock in r: data = remote_sock.recv(4096) if not data: break client_sock.sendall(data) except Exception as e: print(f"[!] 异常: {e}") finally: try: client_sock.close() except: pass if remote_sock: try: remote_sock.close() except: passdef main(): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("0.0.0.0", PORT)) server.listen(128) print(f"====== HTTP 代理服务已启动,监听端口 {PORT} ======") while True: client_sock, addr = server.accept() threading.Thread(target=handle_client, args=(client_sock, addr), daemon=True).start()if __name__ == '__main__': main()
sock5版本启动文件
import socket
import threading
import select
import os
PORT = int(os.environ.get('SERVER_PORT', os.environ.get('PORT', 20006)))# ================= 账号密码配置 =================AUTH_USER = "admin"AUTH_PASS = "123456"# ===============================================def handle_client(client_sock, client_addr): print(f"[+] 收到客户端连接: {client_addr}") remote_sock = None try: # 1. SOCKS5 握手阶段 1: 协商认证方式 header = client_sock.recv(2) if len(header) < 2 or header[0] != 0x05: client_sock.close() return nmethods = header[1] methods = client_sock.recv(nmethods) if 0x02 in methods: client_sock.sendall(b"\x05\x02") # 2. SOCKS5 用户名密码认证子协议 (RFC 1929) sub_ver = client_sock.recv(1) if not sub_ver or sub_ver[0] != 0x01: client_sock.close() return ulen = client_sock.recv(1)[0] user = client_sock.recv(ulen).decode('utf-8', errors='ignore') plen = client_sock.recv(1)[0] passwd = client_sock.recv(plen).decode('utf-8', errors='ignore') if user == AUTH_USER and passwd == AUTH_PASS: client_sock.sendall(b"\x01\x00") else: client_sock.sendall(b"\x01\x01") client_sock.close() print(f"[-] SOCKS5 认证失败: user={user}") return else: client_sock.sendall(b"\x05\x00") # 3. SOCKS5 请求阶段:解析目标地址 req = client_sock.recv(4) if len(req) < 4 or req[0] != 0x05 or req[1] != 0x01: client_sock.close() return atyp = req[3] if atyp == 0x01: target_host = socket.inet_ntoa(client_sock.recv(4)) elif atyp == 0x03: addr_len = client_sock.recv(1)[0] target_host = client_sock.recv(addr_len).decode('utf-8', errors='ignore') elif atyp == 0x04: target_host = socket.inet_ntop(socket.AF_INET6, client_sock.recv(16)) else: client_sock.close() return port_bytes = client_sock.recv(2) target_port = int.from_bytes(port_bytes, 'big') print(f"[->] 正在通过 SOCKS5 转发到目标: {target_host}:{target_port}") remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) remote_sock.connect((target_host, target_port)) reply = b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00" client_sock.sendall(reply) sockets = [client_sock, remote_sock] while True: r, w, e = select.select(sockets, [], [], 60) if not r: break if client_sock in r: data = client_sock.recv(4096) if not data: break remote_sock.sendall(data) elif remote_sock in r: data = remote_sock.recv(4096) if not data: break client_sock.sendall(data) except Exception as e: print(f"[!] 异常: {e}") finally: try: client_sock.close() except: pass if remote_sock: try: remote_sock.close() except: passdef main(): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("0.0.0.0", PORT)) server.listen(128) print(f"====== SOCKS5 代理服务已启动,监听端口 {PORT} ======") while True: client_sock, addr = server.accept() threading.Thread(target=handle_client, args=(client_sock, addr), daemon=True).start()if __name__ == '__main__': main()
正式使用更换文件的账号密码,默认http和sock5账号都是admin,密码:123456
领取的免费容器需要3天续期一次。
https://github.com/zv201413/katabump
保活项目如上