320 lines
9.3 KiB
Python
320 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
USERLENGTH = 50
|
|
|
|
import ipaddress
|
|
import random
|
|
import sys
|
|
import etcd
|
|
import json
|
|
import datetime
|
|
|
|
from flask import Flask, abort, request, Response
|
|
from flask_restful import reqparse
|
|
|
|
def get_random_ip(network):
|
|
net = ipaddress.IPv6Network(network)
|
|
addr_offset = random.randrange(2**64)
|
|
addr = net[0] + addr_offset
|
|
|
|
return addr
|
|
|
|
def require_args(*args):
|
|
parser = reqparse.RequestParser()
|
|
for arg in args:
|
|
parser.add_argument(arg, required=True)
|
|
return parser.parse_args()
|
|
|
|
class Challenge(object):
|
|
""" A sample challenge -- inherit this and overwrite accordingly """
|
|
|
|
points = 0
|
|
provides = []
|
|
requires = []
|
|
description = None
|
|
|
|
def __init__(self, etcdclient):
|
|
self.client = etcdclient
|
|
|
|
def game(self):
|
|
# Check if preconditions are met - otherwise error out
|
|
# This might need to me moved to the Game class to
|
|
# retrieve other challenges that we can recommend
|
|
# for req in self.requires:
|
|
# path = "/ungleichgame/v1/user/{}/{}".format(user, req)
|
|
# try:
|
|
# data = self.client.read(path)
|
|
# except etcd.EtcdKeyNotFound:
|
|
# return None
|
|
|
|
if request.method == 'GET':
|
|
return self.describe()
|
|
if request.method == 'POST':
|
|
return self.solve()
|
|
|
|
def describe(self):
|
|
return self.description
|
|
|
|
def save_points(self, user):
|
|
""" should be called when the challenge was solved successfully"""
|
|
|
|
key = "points/{}".format(user, type(self).__name__)
|
|
self.set_user_key(user, key, self.points)
|
|
|
|
def set_user_key(self, user, key, value):
|
|
path = "/ungleichgame/v1/user/{}/{}".format(user, key)
|
|
self.client.write(path, value)
|
|
|
|
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.
|
|
"""
|
|
def solve(self):
|
|
args = require_args("user", "network")
|
|
network = args['network']
|
|
user = args['user']
|
|
|
|
try:
|
|
net = ipaddress.IPv6Network(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.set_user_key(user, "network", network)
|
|
self.save_points(user)
|
|
|
|
return "Network {} registered, have fun with the next challenge!".format(network)
|
|
|
|
class IPv6Address(Challenge):
|
|
points = 20
|
|
requires = [ "network" ]
|
|
|
|
description = """
|
|
You have setup your network, great!
|
|
Now it is time to show that you are really controlling your network!
|
|
|
|
"""
|
|
|
|
def describe(self):
|
|
pass
|
|
|
|
def solve(self):
|
|
args = require_args("user", "network")
|
|
network = args['network']
|
|
user = args['user']
|
|
|
|
try:
|
|
net = ipaddress.IPv6Network(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.set_user_key(user, "network", network)
|
|
self.save_points(user)
|
|
|
|
return "Network {} registered, have fun with the next challenge!".format(network)
|
|
|
|
|
|
class Game(object):
|
|
def __init__(self, name, etcdclient, etcbase="/ungleichgame/v1"):
|
|
self.client = etcdclient
|
|
self.app = Flask(name)
|
|
|
|
self.app.add_url_rule('/', 'index', self.index)
|
|
self.app.add_url_rule('/points', 'points', self.points)
|
|
self.app.add_url_rule('/register', 'register', self.register, methods=['POST'])
|
|
|
|
# etcd paths are below here
|
|
self.etcbase = etcbase
|
|
self.userbase = "{}/user".format(self.etcbase)
|
|
|
|
# Automate this
|
|
challenges = [ RegisterNet ]
|
|
|
|
self.app.add_url_rule('/challenge', 'list_of_challenges', self.list_of_challenges)
|
|
self.app.add_url_rule('/challenge/', 'list_of_challenges', self.list_of_challenges)
|
|
|
|
self.challenge_names = []
|
|
for challenge in challenges:
|
|
c = challenge(self.client)
|
|
name = type(c).__name__
|
|
self.challenge_names.append(name)
|
|
path = "/challenge/{}".format(name)
|
|
|
|
self.app.add_url_rule(path, name, c.game, methods=['GET', 'POST'])
|
|
|
|
|
|
def list_of_challenges(self):
|
|
base = request.base_url
|
|
challenges = [ "{} ({})".format(name, "{}/{}".format(base, name)) for name in self.challenge_names ]
|
|
|
|
return """The following challenges are available on this server:
|
|
|
|
{}
|
|
|
|
""".format("\n".join(challenges))
|
|
|
|
|
|
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
|
|
|
|
def get_points(self):
|
|
""" Returns a dict['username'] = points """
|
|
|
|
user_points = {}
|
|
|
|
path = "{}/".format(self.userbase)
|
|
users = self.client.get(path)
|
|
|
|
if users:
|
|
print(users)
|
|
for user in users.children:
|
|
username= user.key # needs to be FIXED
|
|
user_points[username] = 0
|
|
|
|
point_path = "{}/points".format(user.key)
|
|
points = self.read_etcd(point_path, recursive=True)
|
|
|
|
if not points:
|
|
continue
|
|
|
|
for challenge in points.children:
|
|
user_points[username] += int(challenge.value)
|
|
|
|
return user_points
|
|
|
|
def index(self):
|
|
points = self.points()
|
|
|
|
return """Welcome to the game server!
|
|
|
|
Current point list is:
|
|
|
|
{}
|
|
|
|
For more information visit
|
|
|
|
https://code.ungleich.ch/nico/ungleich-game
|
|
""".format(points)
|
|
|
|
def points(self):
|
|
point_list = self.get_points()
|
|
res = []
|
|
if not point_list:
|
|
return Response("No winners yet!")
|
|
|
|
for k, v in point_list.items():
|
|
res.append("{} has {} points".format(k, v))
|
|
|
|
return """
|
|
Point list (aka high score)
|
|
---------------------------
|
|
{}
|
|
""".format("\n".join(res))
|
|
|
|
def register(self):
|
|
args = require_args("user")
|
|
path = "{}/{}/registered_at".format(self.userbase, args['user'])
|
|
value = str(datetime.datetime.now())
|
|
|
|
cur = self.read_etcd(path)
|
|
|
|
if cur:
|
|
value = cur.value
|
|
else:
|
|
self.client.write(path, value)
|
|
|
|
return "Registered at: {}\n".format(value)
|
|
|
|
# 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")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# 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')
|