160 lines
5.4 KiB
Python
160 lines
5.4 KiB
Python
import socketserver
|
|
import signal
|
|
import random
|
|
import os
|
|
|
|
|
|
class Task(socketserver.BaseRequestHandler):
|
|
def _recvall(self):
|
|
BUFF_SIZE = 2048
|
|
data = b''
|
|
while True:
|
|
part = self.request.recv(BUFF_SIZE)
|
|
data += part
|
|
if len(part) < BUFF_SIZE:
|
|
break
|
|
return data.strip()
|
|
|
|
def send(self, msg, newline=True):
|
|
try:
|
|
if newline:
|
|
msg += b'\n'
|
|
self.request.sendall(msg)
|
|
except:
|
|
pass
|
|
|
|
def recv(self, prompt=b'[-] '):
|
|
self.send(prompt, newline=False)
|
|
return self._recvall()
|
|
|
|
def select_maze(self, seed:int):
|
|
# 11 * 11 的maze
|
|
if seed % 3 == 0:
|
|
maze = [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
|
|
[1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1],
|
|
[1, 1, 1, 1, 1, 0, 1, 1, 1, 1, 1],
|
|
[1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1],
|
|
[1, 1, 1, 0, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 0, 0, 0, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 1, 0, 1, 1, 1, 1, 1],
|
|
[1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1],
|
|
[1, 0, 1, 0, 1, 1, 1, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 1, 0, 0, 0, 1, 0, 1],
|
|
[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]
|
|
start = [1, 1]
|
|
end = [9, 9]
|
|
# ddddssssssddddss
|
|
elif seed % 3 == 1:
|
|
maze = [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
|
|
[1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1],
|
|
[1, 0, 1, 0, 1, 1, 1, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1],
|
|
[1, 0, 1, 1, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 0, 0, 0, 1, 0, 0, 1],
|
|
[1, 0, 1, 1, 1, 1, 0, 1, 1, 0, 1],
|
|
[1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]
|
|
start = [9, 3]
|
|
end = [9, 5]
|
|
# waawwwwwwwddssddsssdssas
|
|
else:
|
|
maze = [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
|
|
[1, 0, 0, 0, 1, 0, 1, 0, 0, 0, 1],
|
|
[1, 1, 1, 0, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 0, 0, 0, 0, 0, 0, 0, 1, 0, 1],
|
|
[1, 0, 1, 1, 1, 1, 1, 0, 1, 0, 1],
|
|
[1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1],
|
|
[1, 0, 1, 1, 1, 0, 1, 1, 1, 1, 1],
|
|
[1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1],
|
|
[1, 0, 1, 0, 1, 0, 0, 0, 1, 0, 1],
|
|
[1, 0, 0, 0, 1, 1, 1, 0, 0, 0, 1],
|
|
[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]
|
|
start = [1, 9]
|
|
end = [7, 9]
|
|
# aassaaaaaassddddsssddsddww
|
|
return maze, start, end
|
|
|
|
def check_position(self,maze:list,position:list)->bool:
|
|
if maze[position[0]][position[1]] == 1:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def game(self):
|
|
self.send(b"Try to solve this maze")
|
|
self.send(b"Use wasd to move")
|
|
maze, start, end = self.select_maze(random.randint(0,99))
|
|
position = start
|
|
|
|
while position != end:
|
|
|
|
move = self.recv(b'Try to move: ').decode()
|
|
if not move:
|
|
print("Connection closed by the client.")
|
|
return False
|
|
|
|
if len(move) != 1:
|
|
self.send(b'Invalid lenth')
|
|
continue
|
|
if move not in 'wasd':
|
|
self.send(b'Invalid instruction')
|
|
continue
|
|
if move == "w":
|
|
position[0] -= 1
|
|
if self.check_position(maze,position):
|
|
self.send(b"You moved one step up")
|
|
else:
|
|
self.send(b"You bumped into the wall")
|
|
position[0] += 1
|
|
elif move == "s":
|
|
position[0] += 1
|
|
if self.check_position(maze,position):
|
|
self.send(b"You moved one step down")
|
|
else:
|
|
self.send(b"You bumped into the wall")
|
|
position[0] -= 1
|
|
elif move == "a":
|
|
position[1] -= 1
|
|
if self.check_position(maze,position):
|
|
self.send(b"You moved one step left")
|
|
else:
|
|
self.send(b"You bumped into the wall")
|
|
position[1] += 1
|
|
else:
|
|
position[1] += 1
|
|
if self.check_position(maze,position):
|
|
self.send(b"You moved one step right")
|
|
else:
|
|
self.send(b"You bumped into the wall")
|
|
position[1] -= 1
|
|
return True
|
|
|
|
|
|
def handle(self):
|
|
signal.alarm(60) # type: ignore
|
|
if not self.game():
|
|
self.send(b'[!] Wrong!')
|
|
return
|
|
|
|
self.send(b'here is your flag')
|
|
self.send(flag)
|
|
|
|
|
|
class ThreadedServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
|
|
pass
|
|
|
|
|
|
class ForkedServer(socketserver.ForkingMixIn, socketserver.TCPServer):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
flag = bytes(os.getenv("FLAG"),"utf-8")
|
|
# flag = b'11111111111111'
|
|
HOST, PORT = '0.0.0.0', 10001
|
|
server = ForkedServer((HOST, PORT), Task)
|
|
server.allow_reuse_address = True
|
|
print(HOST, PORT)
|
|
server.serve_forever() |