ungleich-game/server.py

230 lines
6.9 KiB
Python
Raw Normal View History

2019-05-26 19:19:58 +00:00
#!/usr/bin/env python3
2019-04-14 16:57:39 +00:00
2019-05-26 19:19:58 +00:00
USERLENGTH = 50
2019-04-14 16:57:39 +00:00
2019-05-26 19:19:58 +00:00
import ipaddress
import random
import sys
import etcd
import ungleichapi
import json
import datetime
2019-04-14 16:57:39 +00:00
2019-05-26 19:19:58 +00:00
from flask import Flask, abort, request, Response
from flask_restful import reqparse
2019-04-14 16:57:39 +00:00
2019-05-26 19:19:58 +00:00
# app = Flask(__name__)
2019-04-14 16:57:39 +00:00
2019-05-26 19:19:58 +00:00
def get_random_ip(network):
net = ipaddress.IPv6Network(network)
addr_offset = random.randrange(2**64)
addr = net[0] + addr_offset
2019-04-14 16:57:39 +00:00
2019-05-26 19:19:58 +00:00
return addr
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
class Challenge(object):
""" A sample challenge -- inherit this and overwrite accordingly """
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
points = 0
provides = []
requires = []
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
def __init__(self, etcdclient):
self.client = etcdclient
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
def require_args(self, *args):
parser = reqparse.RequestParser()
for arg in args:
parser.add_argument(arg, required=True)
return parser.parse_args()
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
def describe(self):
return self.description
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
def save_points(self, user):
""" should be called when the challenge was solved successfully"""
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
path = "/ungleichgame/v1/{}/challenges/{}/points".format(user, self.__name__)
self.client.write(path, self.points)
def solve(self):
""" Needs to be implemented per challenge """
pass
class RegisterNet(Challenge):
points = 10
provides = [ "network" ]
description = """
Register a /64 IPv6 network that you fully control.
Many other challenges depend on this. You will need to
be able to configure IPv6 addresses in this networks
and to setup services listening on these IPv6 addresses.
Submit your network with the "network" parameter.
2019-05-11 22:18:03 +00:00
"""
2019-05-26 19:19:58 +00:00
def solve(self):
self.require_args("user", "network")
try:
net = ipaddress.IPv6Network(args['network'])
network = args['network']
except Exception as e:
return Response(status=400, response="Cannot register network {}: {}".format(network, e))
if not net.prefixlen == 64:
return Response(status=400, response="{} mask is not /64 - please use a /64 network".format(net))
# Save network
self.client.write("/ungleichgame/v1/{}/network".format(user), network)
self.save_points(args['user'])
return json.dumps("All good, go to /level/1 to start with level 1! - {}".format(data.value))
class Game(object):
def __init__(self, name, etcdclient, etcbase="/ungleichgame/v1"):
self.client = etcdclient
self.app = Flask(name)
self.app.add_url_rule('/', 'highscore', self.highscore)
self.app.add_url_rule('/highscore', 'highscore', self.highscore)
# etcd paths are below here
self.etcbase = etcbase
self.userbase = "{}/user".format(self.etcbase)
# Automate this
challenges = [ RegisterNet ]
for challenge in challenges:
c = challenge(self.client)
name = c.__name__
path = "/challenge/{}".format(name)
self.app.add_url_rule(path, name, c.describe, methods=['GET'])
self.app.add_url_rule(path, name, c.solve, methods=['POST'])
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
def read_etcd(self, path, recursive=False):
try:
data = self.client.read(path, recursive=recursive)
except etcd.EtcdKeyNotFound:
return None
except Exception:
abort(Response(status=400, response="Error connecting to etcd"))
return data
2019-05-11 22:18:03 +00:00
2019-05-26 19:19:58 +00:00
def get_highscore(self, username=None):
""" Returns a dict['username'] = points """
all_users = {}
highscore = {}
print("getting high")
if username:
path = "{}/{}".format(self.userbase, username)
user = self.read_etcd(path)
if user:
all_users[username] = user
else:
path = "{}/".format(self.userbase)
users = self.read_etcd(path, recursive=True)
print("reading from {}".format(path))
if users:
for child in users.children:
print("adding user {} {} = {}".format(child, child.key, child.value))
all_users[child.key] = child.value
for k, v in all_users.items():
# Ignore all kind of errors - just add the ones that work
try:
highscore[k] = json.loads(v)['points']
print("f?")
except Exception as e:
print(e)
return highscore
def highscore(self):
point_list = self.get_highscore()
res = []
if not point_list:
return Response("No winners yet!")
for k, v in point_list.items():
res.append("<p>{} has {} points</p>".format(k, v))
return Response("\n".join(res))
# def get_ip_address():
# args = self.require_args("network", "user")
# # Needs to be fixed with ungleich-otp
# username=args['user']
# if request.method == 'GET':
# return Response("""
# This is an easy level - just register any /64 network
# that you fully control. After submission the game server will generate
# a random IPv6 address in this network.
# """)
# client = etcd.Client(port=2379)
# try:
# data = client.read("/ungleichgame/v1/{}/network".format(username))
# # FIXME: differentiate keynotfound and other errors
# except Exception as e:
# return Response(status=400, response="Cannot read your network, try registering first (error: {})".format(e))
# return Response("data={}".format(data.value))
# address = get_random_ip(data.value)
# # FIXME: catch errors
# client.write("/ungleichgame/v1/{}/address".format(username), address)
# return Response("Your IPv6 address for this game is {}. Make it pingable and post to /level/1/result".format(address))
# @app.route("/level/2", methods=['GET', 'POST']) # post for username
# def pingme():
# parser = reqparse.RequestParser()
# parser.add_argument('user', required=True)
# args = parser.parse_args()
# # Needs to be fixed with ungleich-otp
# username=args['user']
# if request.method == 'GET':
# return Response("""
# Proof that you can really control the network that you submitted:
# - Setup the IPv6 address to be ping6 able globally
# - POST to this address when it is configured
# """)
# if request.method == 'POST':
# try:
# data = client.read("/ungleichgame/v1/{}/address".format(username), address)
# except Exception as e:
# return Response(status=400,
# response="""
# You need to register a network before trying to be reachable.
# Please go back to Level 1 for registering your network.
# """)
# return Response("something good")
2019-04-14 16:57:39 +00:00
if __name__ == '__main__':
2019-05-26 19:19:58 +00:00
# net_base = "2a0a:e5c1:{:x}::/64"
# net_offset = random.randrange(0xffff)
# net = ipaddress.IPv6Network(net_base.format(net_offset))
# username = 'nico{}'.format(net_offset)
# print("{} has {}".format(username, net))
g = Game(__name__, etcd.Client(port=2379))
g.app.run(host="::", port='5002')