mirror of
https://github.com/MoHadiShibli/Control4Free.git
synced 2026-10-06 12:00:30 +02:00
MoHadiShibli
baf36d648e
A game talks to a virtual controller as it would to a real one, and the system
keeps what it asked for. scePadVirtualDeviceGetRemoteSetting reads it back in
the layout of a DualShock 4's output report: the two motors at [3] and [4], the
light bar at [5..7]. The layout was read on the console while games rumbled and
users signed in, and feedback.c is tested against those exact buffers.
The service reads it every 16 ms for each controller; games pulse rumble for
only tens of milliseconds, which a 100 ms poll missed.
- Rumble goes to the controller's owner as {"method":"v","params":[pad,large,
small]}, once per change, and to whoever takes a controller over, so a page
never keeps buzzing on old news. The page already drove an Android phone's
vibration and a gamepad's actuators from it.
- The light bar goes into the status everyone sees. The PS4 sets player colours
at a quarter strength, so the colour is brightened to full, keeping its hue;
an unlit bar keeps the controller's own colour.
- Without the call, controllers work as before and it is logged once.
Also: the page says "Choose user on PS4", not "on TV", since not everyone plays
on a TV. The test stub takes C4F-TEST-FEEDBACK lines to set what the "game"
wants, and the test client can wait for messages the service pushes.
443 lines
20 KiB
Python
443 lines
20 KiB
Python
"""Real transport + application tests; VDA calls are replaced by web_stub.c."""
|
|
import base64
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BINARY = ROOT / 'build/web-host-test'
|
|
|
|
|
|
def build():
|
|
subprocess.run(['python3', 'tools/embed_client.py', 'client/index.html', 'build/client.c'], check=True)
|
|
subprocess.run(['python3', 'tools/embed_file.py', 'build/assets.c',
|
|
'c4fManifest=client/manifest.webmanifest', 'c4fIcon=client/icon-192.png'], check=True)
|
|
subprocess.run(['clang-18', '-std=gnu11', '-Wall', '-Wextra', '-Werror', '-g',
|
|
'-Iinclude', '-Ivendor/jsmn', '-DC4F_STALE_MS=400', '-DC4F_RELEASE_MS=1800',
|
|
'src/net.c', 'src/web.c', 'src/klog_line.c', 'tests/web_stub.c', 'build/client.c',
|
|
'build/assets.c', '-pthread', '-o', str(BINARY)], check=True)
|
|
|
|
|
|
class Client:
|
|
def __init__(self, port=4264):
|
|
self.sock = socket.create_connection(('127.0.0.1', port), timeout=3)
|
|
self.buf = b''
|
|
self.pushed = [] # messages the service sent on its own, oldest first
|
|
key = 'dGhlIHNhbXBsZSBub25jZQ=='
|
|
self.sock.sendall((f'GET /ws HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nOrigin: http://127.0.0.1:{port}\r\n'
|
|
f'Connection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n').encode())
|
|
while b'\r\n\r\n' not in self.buf:
|
|
self.buf += self.sock.recv(8192)
|
|
head, self.buf = self.buf.split(b'\r\n\r\n', 1)
|
|
assert b'101 Switching' in head, head
|
|
assert b's3pPLMBiTxaQ9kYGzzhZRbK+xOo=' in head, head
|
|
self.next_id = 1
|
|
|
|
def exact(self, size):
|
|
while len(self.buf) < size:
|
|
chunk = self.sock.recv(8192)
|
|
assert chunk, 'unexpected close'
|
|
self.buf += chunk
|
|
result, self.buf = self.buf[:size], self.buf[size:]
|
|
return result
|
|
|
|
def frame(self, payload, op=1, fin=True, masked=True):
|
|
if isinstance(payload, str):
|
|
payload = payload.encode()
|
|
mask = b'abcd'
|
|
head = bytes([(128 if fin else 0) | op, (128 if masked else 0) | (len(payload) if len(payload) < 126 else 126)])
|
|
if len(payload) >= 126:
|
|
head += struct.pack('!H', len(payload))
|
|
self.sock.sendall(head + (mask if masked else b'') + bytes(b ^ (mask[i % 4] if masked else 0) for i, b in enumerate(payload)))
|
|
|
|
def receive(self):
|
|
first, second = self.exact(2)
|
|
size = second & 127
|
|
if size == 126:
|
|
size = struct.unpack('!H', self.exact(2))[0]
|
|
assert not second & 128
|
|
return first & 15, self.exact(size)
|
|
|
|
def request(self, method, params=()):
|
|
ident = self.next_id
|
|
self.next_id += 1
|
|
self.frame(json.dumps(dict(id=ident, method=method, params=list(params))))
|
|
while True:
|
|
op, text = self.receive()
|
|
if op != 1:
|
|
continue
|
|
msg = json.loads(text)
|
|
if msg.get('id') == ident:
|
|
return msg
|
|
if 'id' not in msg:
|
|
self.pushed.append(msg)
|
|
|
|
def wait_for(self, match, timeout=2):
|
|
"""The first pushed message `match` accepts, seen already or still to come."""
|
|
end = time.monotonic() + timeout
|
|
while True:
|
|
for i, msg in enumerate(self.pushed):
|
|
if match(msg):
|
|
del self.pushed[:i + 1]
|
|
return msg
|
|
self.pushed.clear()
|
|
assert time.monotonic() < end, 'the expected message never came'
|
|
op, text = self.receive()
|
|
if op == 1:
|
|
msg = json.loads(text)
|
|
if 'id' not in msg:
|
|
self.pushed.append(msg)
|
|
|
|
def input(self, pad, buttons=0, lx=128, ly=128, rx=128, ry=128, l2=0, r2=0, touch=()):
|
|
params = [pad, buttons, lx, ly, rx, ry, l2, r2, len(touch)]
|
|
for t in touch:
|
|
params.extend(t)
|
|
self.frame(json.dumps(dict(method='u', params=params)))
|
|
|
|
def close(self):
|
|
self.sock.close()
|
|
|
|
|
|
class Server:
|
|
def __init__(self, **options):
|
|
readfd, self.log = os.pipe()
|
|
os.set_blocking(readfd, False)
|
|
self.trace = tempfile.TemporaryFile(mode='w+')
|
|
env = dict(os.environ, **{'C4F_TEST_KLOG_FD': str(readfd), **options})
|
|
self.p = subprocess.Popen([str(BINARY)], env=env, pass_fds=(readfd,), stdout=self.trace)
|
|
os.close(readfd)
|
|
for _ in range(100):
|
|
try:
|
|
self.c = Client()
|
|
break
|
|
except ConnectionRefusedError:
|
|
time.sleep(.02)
|
|
else:
|
|
raise AssertionError('server not listening')
|
|
|
|
def rows(self):
|
|
self.trace.seek(0)
|
|
return self.trace.read().splitlines()
|
|
|
|
def close(self):
|
|
self.c.close()
|
|
if self.p.poll() is None:
|
|
self.p.terminate()
|
|
self.p.wait(timeout=3)
|
|
if self.log is not None:
|
|
os.close(self.log)
|
|
self.trace.close()
|
|
|
|
|
|
def http(request):
|
|
with socket.create_connection(('127.0.0.1', 4264), timeout=3) as s:
|
|
s.sendall(request)
|
|
parts = []
|
|
while chunk := s.recv(65536):
|
|
parts.append(chunk)
|
|
return b''.join(parts)
|
|
|
|
|
|
def main():
|
|
build()
|
|
server = Server()
|
|
try:
|
|
response = http(b'GET / HTTP/1.1\r\nHost: 127.0.0.1:4264\r\n\r\n')
|
|
headers, body = response.split(b'\r\n\r\n', 1)
|
|
assert b'200 OK' in headers and b'Content-Security-Policy:' in headers
|
|
assert gzip.decompress(body) == Path('client/index.html').read_bytes()
|
|
assert not any(r.startswith('ADD') for r in server.rows()), 'page load created a controller'
|
|
assert b'403' in http(b'GET / HTTP/1.1\r\nHost: evil.example:4264\r\n\r\n')
|
|
assert b'403' in http(b'GET /ws HTTP/1.1\r\nHost: 127.0.0.1:4264\r\nOrigin: https://evil.example\r\nUpgrade: websocket\r\n\r\n')
|
|
assert b'404' in http(b'GET /../README.md HTTP/1.1\r\nHost: 127.0.0.1:4264\r\n\r\n')
|
|
print('PASS embedded page, RFC handshake, origin/host/path checks, no automatic creation', flush=True)
|
|
|
|
# Keeping the page on a phone's home screen needs these two files.
|
|
headers, body = http(b'GET /manifest.webmanifest HTTP/1.1\r\nHost: 127.0.0.1:4264\r\n\r\n').split(b'\r\n\r\n', 1)
|
|
assert b'200 OK' in headers and b'Content-Type: application/manifest+json' in headers
|
|
manifest = json.loads(body)
|
|
assert manifest['start_url'] == '/' and manifest['icons'][0]['src'] == '/icon-192.png'
|
|
assert body == Path('client/manifest.webmanifest').read_bytes()
|
|
headers, body = http(b'GET /icon-192.png HTTP/1.1\r\nHost: 127.0.0.1:4264\r\n\r\n').split(b'\r\n\r\n', 1)
|
|
assert b'200 OK' in headers and b'Content-Type: image/png' in headers
|
|
assert body == Path('client/icon-192.png').read_bytes() and body[:8] == b'\x89PNG\r\n\x1a\n'
|
|
page = Path('client/index.html').read_text()
|
|
assert 'href="/manifest.webmanifest"' in page and 'href="/icon-192.png"' in page
|
|
assert b'403' in http(b'GET /icon-192.png HTTP/1.1\r\nHost: evil.example:4264\r\n\r\n')
|
|
print('PASS manifest and icon served for a home-screen shortcut', flush=True)
|
|
|
|
c = server.c
|
|
assert c.request('info')['result']['pads'] == 4
|
|
c.frame(b'ping', op=9)
|
|
while True:
|
|
op, data = c.receive()
|
|
if op == 10:
|
|
assert data == b'ping'
|
|
break
|
|
c.frame('{"id":99,"method":', fin=False)
|
|
c.frame('"status","params":[]}', op=0)
|
|
while True:
|
|
_, data = c.receive()
|
|
msg = json.loads(data)
|
|
if msg.get('id') == 99:
|
|
assert msg['result']['pads'][0]['state'] == 'free'
|
|
break
|
|
assert 'error' in c.request('claim', [4])
|
|
assert c.request('claim', [0])['result']['pads'][0]['state'] == 'select'
|
|
assert all(row.split()[3] == '0' for row in server.rows() if row.startswith('FRAME')), 'creation injected a button'
|
|
other = Client()
|
|
assert other.request('claim', [0])['error']['code'] == 409
|
|
assert c.request('claim', [0, 1, 2, 3])['result']['pads'][3]['mine']
|
|
print('PASS ping, fragmented messages, explicit creation, four slots, exclusive ownership', flush=True)
|
|
|
|
# Exact device event determines assignment; no generic physical-pad read handle.
|
|
os.write(server.log, b'<118>DEVICE_OWNER_CHANGED [DeviceId:0x11030d][UserId:0x1a2b3c4d]\n')
|
|
time.sleep(.04)
|
|
state = c.request('status')['result']['pads']
|
|
assert state[0]['state'] == 'ready' and state[1]['state'] == 'select'
|
|
c.input(0, 0x4000, lx=1, ry=254, l2=12, r2=230, touch=[(3, 1000, 500)])
|
|
c.input(0)
|
|
c.input(1, 0x10000)
|
|
time.sleep(.10)
|
|
frames = [r.split() for r in server.rows() if r.startswith('FRAME')]
|
|
assert any(r[2:11] == ['11030d', '16384', '1', '128', '128', '254', '12', '230', '1'] for r in frames)
|
|
assert any(r[2] == '12030d' and r[3] == '65536' for r in frames)
|
|
assert not any(r[2] in ['13030d', '14030d'] and r[3] != '0' for r in frames)
|
|
assert any(r[2] == '11030d' and r[3] == '0' for r in frames[-30:])
|
|
print('PASS device-specific assignment, PS/Cross, axes/triggers/touch, button edges, pad isolation', flush=True)
|
|
|
|
time.sleep(.5)
|
|
assert c.request('status')['result']['pads'][1]['state'] == 'paused'
|
|
frames = [r.split() for r in server.rows() if r.startswith('FRAME')]
|
|
assert all(r[3] == '0' for r in frames[-12:]), 'stale input stayed pressed'
|
|
c.input(1, 0x20)
|
|
time.sleep(.04)
|
|
assert c.request('status')['result']['pads'][1]['state'] == 'select'
|
|
assert other.request('stop')['error']['code'] == 409
|
|
other.close()
|
|
assert not c.request('claim', [])['result']['pads'][0]['open']
|
|
assert len([r for r in server.rows() if r.startswith('REMOVE')]) == 4
|
|
print('PASS idle neutralization, recovery, release and stopping protection', flush=True)
|
|
|
|
assert c.request('claim', [0])['result']['pads'][0]['open']
|
|
c.input(0, 0x4000)
|
|
time.sleep(.03)
|
|
c.close()
|
|
time.sleep(.08)
|
|
c = server.c = Client()
|
|
assert c.request('status')['result']['pads'][0]['state'] == 'paused'
|
|
assert c.request('claim', [0])['result']['pads'][0]['mine']
|
|
time.sleep(1.9)
|
|
assert c.request('status')['result']['pads'][0]['state'] == 'free'
|
|
c.frame(json.dumps(dict(method='stop', params=[])))
|
|
assert server.p.wait(timeout=3) == 0
|
|
print('PASS disconnect releases buttons, reconnect grace and inactive-device cleanup', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
server = Server(C4F_FAIL_ADD='1')
|
|
try:
|
|
assert server.c.request('claim', [0])['error']['code'] == 503
|
|
assert server.c.request('claim', [0])['error']['code'] == 503
|
|
assert server.c.request('status')['result']['pads'][0]['state'] == 'free'
|
|
server.c.frame(json.dumps(dict(method='stop', params=[])))
|
|
assert server.p.wait(timeout=3) == 0
|
|
print('PASS failed device creation is visible and retries do not accumulate orphan devices', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# AutoRun can start the payload before GoldHEN's klog server listens.
|
|
server = Server(C4F_TEST_KLOG_BUSY='1')
|
|
try:
|
|
assert server.c.request('claim', [0])['error']['code'] == 503
|
|
assert 'KLOG none' in server.rows() and not any(r.startswith('ADD') for r in server.rows())
|
|
finally:
|
|
server.close()
|
|
# A reader that opens but delivers nothing is dropped before AddDevice runs.
|
|
server = Server(C4F_TEST_KLOG_DEAD='1')
|
|
try:
|
|
assert server.c.request('claim', [0])['error']['code'] == 503
|
|
assert not any(r.startswith('ADD') for r in server.rows())
|
|
assert 'released klog reader' in server.rows()
|
|
assert server.c.request('info')['result']['pads'] == 4
|
|
finally:
|
|
server.close()
|
|
# A klog source that dies is reopened for the next new controller.
|
|
server = Server()
|
|
try:
|
|
assert server.c.request('claim', [0])['result']['pads'][0]['mine']
|
|
os.write(server.log, b'C4F-TEST-CLOSE-KLOG\n')
|
|
end = time.monotonic() + 3
|
|
while not any('klog source closed' in r for r in server.rows()):
|
|
assert time.monotonic() < end, [r for r in server.rows() if not r.startswith('FRAME')]
|
|
server.c.input(0) # keep the controller, so only the dead log can free the reader
|
|
time.sleep(.02)
|
|
assert server.c.request('claim', [0, 1])['result']['pads'][1]['mine']
|
|
assert len([r for r in server.rows() if r.startswith('KLOG ') and r != 'KLOG none']) == 2
|
|
print('PASS klog unavailable, silent or closed: refused safely, then reopened for the next controller', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# A creation in progress must not stop anyone else being served.
|
|
server = Server(C4F_TEST_ADD_DELAY='900')
|
|
try:
|
|
first = server.c
|
|
assert first.request('claim', [0])['result']['pads'][0]['mine']
|
|
second = Client()
|
|
result = []
|
|
thread = threading.Thread(target=lambda: result.append(second.request('claim', [1])))
|
|
thread.start()
|
|
time.sleep(.3)
|
|
started = time.monotonic()
|
|
state = first.request('status')['result']['pads']
|
|
elapsed = time.monotonic() - started
|
|
assert elapsed < .1, f'status waited {elapsed:.2f}s for the other player'
|
|
assert state[1]['state'] == 'connecting'
|
|
first.input(0, 0x4000)
|
|
time.sleep(.05)
|
|
frames = [r.split() for r in server.rows() if r.startswith('FRAME')]
|
|
assert any(r[2] == '11030d' and r[3] == '16384' for r in frames), 'input stopped during creation'
|
|
thread.join(timeout=5)
|
|
assert result and result[0]['result']['pads'][1]['mine']
|
|
second.close()
|
|
print('PASS a controller being created does not block the other players', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# Input goes out when it arrives, and an idle controller costs almost nothing.
|
|
server = Server()
|
|
try:
|
|
c = server.c
|
|
assert c.request('claim', [0])['result']['pads'][0]['mine']
|
|
assert 'result' in c.request('ping')
|
|
time.sleep(.6) # settle into the keepalive rate
|
|
before = len([r for r in server.rows() if r.startswith('FRAME')])
|
|
c.input(0, 0x4000)
|
|
time.sleep(.02)
|
|
frames = [r.split() for r in server.rows() if r.startswith('FRAME')]
|
|
pressed = next((i for i, r in enumerate(frames) if i >= before and r[3] == '16384'), None)
|
|
assert pressed is not None, 'the press never reported'
|
|
assert pressed - before <= 1, f'the press waited for {pressed - before} keepalive reports'
|
|
# While input is moving: about 250 Hz. Idle: the keepalive rate.
|
|
start = len([r for r in server.rows() if r.startswith('FRAME')])
|
|
for _ in range(10):
|
|
c.input(0, 0x4000)
|
|
time.sleep(.02)
|
|
moving = len([r for r in server.rows() if r.startswith('FRAME')]) - start
|
|
time.sleep(.8)
|
|
start = len([r for r in server.rows() if r.startswith('FRAME')])
|
|
time.sleep(.4)
|
|
idle = len([r for r in server.rows() if r.startswith('FRAME')]) - start
|
|
assert moving >= 30, f'only {moving} reports in 200ms of input'
|
|
assert idle <= 40, f'{idle} reports in 400ms of idling'
|
|
print('PASS input reports on arrival, fast while it moves and slow while it does not', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# A controller must survive the moment it is created. A claim finishes inside a
|
|
# main-loop iteration and stamps its clock after the iteration read its own; on
|
|
# the console the gap is InsertData and klog work, here C4F_TEST_SLOW_MS. Those
|
|
# unsigned stamps compared directly wrapped, and the idle reaper deleted every
|
|
# controller in the claim at once, reporting it as unused (console, 1.0.0 dev).
|
|
server = Server(C4F_TEST_SLOW_MS='3')
|
|
try:
|
|
c = server.c
|
|
for round in range(4):
|
|
assert c.request('claim', [0])['result']['pads'][0]['mine']
|
|
time.sleep(.05)
|
|
state = c.request('status')['result']['pads'][0]
|
|
assert state['open'] and state['mine'], f'round {round}: the new controller went away: {state}'
|
|
# Adding a second one refreshes both: neither may go.
|
|
assert c.request('claim', [0, 1])['result']['pads'][1]['mine']
|
|
time.sleep(.05)
|
|
pads = c.request('status')['result']['pads']
|
|
assert all(pads[i]['open'] and pads[i]['mine'] for i in (0, 1)), f'round {round}: {pads[:2]}'
|
|
c.request('claim', [])
|
|
assert not any('removed after' in r for r in server.rows()), [r for r in server.rows() if 'removed after' in r]
|
|
print('PASS a new controller is not mistaken for an idle one', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# Rumble goes to the controller's owner; the light bar, brightened, to everyone.
|
|
def rumble(pad, large, small):
|
|
return lambda m: m.get('method') == 'v' and m['params'] == [pad, large, small]
|
|
server = Server()
|
|
try:
|
|
c = server.c
|
|
assert c.request('claim', [0])['result']['pads'][0]['mine']
|
|
assert c.request('status')['result']['pads'][0]['color'] == [32, 96, 255], 'unlit: its own colour'
|
|
os.write(server.log, b'C4F-TEST-FEEDBACK 11030d 200 40 64 0 0\n')
|
|
c.wait_for(rumble(0, 200, 40))
|
|
time.sleep(.05)
|
|
assert c.request('status')['result']['pads'][0]['color'] == [255, 0, 0], 'player 2 red, at full'
|
|
watcher = Client()
|
|
os.write(server.log, b'C4F-TEST-FEEDBACK 11030d 0 0 0 64 0\n')
|
|
c.wait_for(rumble(0, 0, 0))
|
|
time.sleep(.05)
|
|
assert watcher.request('status')['result']['pads'][0]['color'] == [0, 255, 0], 'everyone sees it'
|
|
os.write(server.log, b'C4F-TEST-FEEDBACK 11030d 0 0 32 0 32\n')
|
|
time.sleep(.1)
|
|
assert c.request('status')['result']['pads'][0]['color'] == [255, 0, 255], 'pink keeps its hue'
|
|
assert not any(m.get('method') == 'v' for m in watcher.pushed), 'rumble went to a non-owner'
|
|
# Rumble holds while the game holds it; one message per change, not per poll.
|
|
os.write(server.log, b'C4F-TEST-FEEDBACK 11030d 90 0 32 0 32\n')
|
|
c.wait_for(rumble(0, 90, 0))
|
|
time.sleep(.3)
|
|
c.request('status')
|
|
assert not any(m.get('method') == 'v' for m in c.pushed), 'unchanged rumble was sent again'
|
|
# Whoever takes the controller over is told the current state at once.
|
|
c.close()
|
|
server.c = c = Client()
|
|
assert c.request('claim', [0])['result']['pads'][0]['mine']
|
|
c.wait_for(rumble(0, 90, 0))
|
|
# And told when it stops, so a page never keeps buzzing on old news.
|
|
os.write(server.log, b'C4F-TEST-FEEDBACK 11030d 0 0 0 0 0\n')
|
|
c.wait_for(rumble(0, 0, 0))
|
|
time.sleep(.05)
|
|
assert c.request('status')['result']['pads'][0]['color'] == [32, 96, 255], 'unlit again: its own colour'
|
|
watcher.close()
|
|
print('PASS rumble to the owner on change, the light bar to everyone, the state to a new owner', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# Without the call, controllers work as before.
|
|
server = Server(C4F_TEST_NO_FEEDBACK='1')
|
|
try:
|
|
assert server.c.request('claim', [0, 1])['result']['pads'][1]['mine']
|
|
time.sleep(.1)
|
|
assert server.c.request('status')['result']['pads'][0]['color'] == [32, 96, 255]
|
|
assert len([r for r in server.rows() if 'not exported' in r]) == 1
|
|
print('PASS no rumble call: logged once, controllers unaffected', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
# Two claims cannot create at once: the second is refused, not queued.
|
|
server = Server(C4F_TEST_ADD_DELAY='600')
|
|
try:
|
|
first = server.c
|
|
result = []
|
|
thread = threading.Thread(target=lambda: result.append(first.request('claim', [0])))
|
|
thread.start()
|
|
time.sleep(.3)
|
|
second = Client()
|
|
assert second.request('claim', [1])['error']['code'] == 409
|
|
thread.join(timeout=5)
|
|
assert result and result[0]['result']['pads'][0]['mine']
|
|
assert second.request('claim', [1])['result']['pads'][1]['mine']
|
|
second.close()
|
|
print('PASS one controller is created at a time, and the next request works', flush=True)
|
|
finally:
|
|
server.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|