tqcq/small-package
0
0
Fork 0
mirror of https://github.com/kenzok8/small-package.git synced 2026-02-06 14:56:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

update 2024-11-27 14:16:39

Browse Source
This commit is contained in:
kenzok8
2024-11-27 14:16:39 +08:00
parent f1331b5323
commit e7f98ab0e7
2 changed files with 27 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
UA2F/scripts/requirements.txt
View File

@@ -1,4 +1,5 @@
requests requests
fake-useragent fake-useragent
tqdm tqdm
systemd-python fastapi
uvicorn

54
UA2F/scripts/test.py
View File

@@ -9,43 +9,43 @@ import subprocess
import sys import sys
import threading import threading
import time import time
from http.server import HTTPServer, BaseHTTPRequestHandler
import requests import requests
from fake_useragent import UserAgent from fake_useragent import UserAgent
from tqdm import tqdm from tqdm import tqdm
from fastapi import FastAPI, Request
from fastapi.responses import Response
from fastapi.staticfiles import StaticFiles
from uvicorn import Config, Server
ua = UserAgent() ua = UserAgent()
PORT = 37491 PORT = 37491
class MyHandler(http.server.SimpleHTTPRequestHandler): app = FastAPI()
def log_message(self, format, *args):
pass
def do_GET(self): @app.get("/")
user_agent = self.headers.get('User-Agent') async def root(request: Request):
user_agent = request.headers.get("user-agent")
# assert user_agent only contains F if not all(c == 'F' for c in user_agent):
if not all([c == 'F' for c in user_agent]): return Response(status_code=400)
self.send_response(400)
logging.error(f"Invalid User-Agent: {user_agent}")
else:
self.send_response(200)
self.end_headers()
ua_len = len(user_agent)
self.wfile.write(str(ua_len).encode())
return Response(content=str(len(user_agent)).encode())
def start_server(): def start_server():
with socketserver.TCPServer(('', PORT), MyHandler, bind_and_activate=False) as httpd: config4 = Config(app=app, host="127.0.0.1", port=PORT, access_log=False)
httpd.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) config6 = Config(app=app, host="::1", port=PORT, access_log=False)
httpd.server_bind() server4 = Server(config4)
httpd.server_activate() server6 = Server(config6)
print(f"Serving on port {PORT}") t4 = threading.Thread(target=server4.run)
httpd.serve_forever() t4.daemon = True
atexit.register(httpd.shutdown) t6 = threading.Thread(target=server6.run)
t6.daemon = True
t4.start()
t6.start()
def start_ua2f(u: str): def start_ua2f(u: str):
p = subprocess.Popen([u]) p = subprocess.Popen([u])
@@ -73,11 +73,7 @@ if __name__ == "__main__":
setup_iptables() setup_iptables()
server_thread = threading.Thread(target=start_server) start_server()
server_thread.daemon = True
server_thread.start()
print(f"Starting server on port {PORT}")
ua2f_thread = threading.Thread(target=start_ua2f, args=(ua2f,)) ua2f_thread = threading.Thread(target=start_ua2f, args=(ua2f,))
ua2f_thread.daemon = True ua2f_thread.daemon = True
@@ -87,7 +83,7 @@ if __name__ == "__main__":
time.sleep(3) time.sleep(3)
for i in tqdm(range(2000)): for i in tqdm(range(1024)):
nxt = ua.random nxt = ua.random
response = requests.get(f"http://127.0.0.1:{PORT}", headers={ response = requests.get(f"http://127.0.0.1:{PORT}", headers={
"User-Agent": nxt "User-Agent": nxt
@@ -95,7 +91,7 @@ if __name__ == "__main__":
assert response.ok assert response.ok
assert response.text == str(len(nxt)) assert response.text == str(len(nxt))
for i in tqdm(range(2000)): for i in tqdm(range(4096)):
nxt = ua.random nxt = ua.random
response = requests.get(f"http://[::1]:{PORT}", headers={ response = requests.get(f"http://[::1]:{PORT}", headers={
"User-Agent": nxt "User-Agent": nxt