1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
from flask import Blueprint, abort, render_template, request, flash, redirect, url_for
from flask_login import login_required, current_user
import ulid
import flask
import ipaddress
from datetime import datetime, timedelta
from app import db
from app.database import EnrollRequest, Network, Peer
from .forms import NewNetworkForm
bp = Blueprint('manage', __name__, url_prefix="/manage")
@bp.route("/networks", methods=["GET", "POST"])
@login_required
def list_networks():
nets = current_user.networks
form = NewNetworkForm(request.form)
if request.method == "POST" and form.validate_on_submit():
subnet = request.form.get('subnet')
description = request.form.get('description')
n = Network(
id=str(ulid.ulid()),
subnet=subnet,
description=description,
manager_id=str(current_user.id)
)
db.session.add(n)
db.session.commit()
flash("Network added")
return render_template("network_list.html", nets=nets, form=form)
@bp.route("/networks/<string:id>/delete")
@login_required
def del_net(id):
n = Network.query.filter_by(id=id).first()
if n.manager_id != current_user.id:
flash("You aren't a manager of this network.")
return redirect(url_for("manage.list_networks"))
db.session.delete(n)
db.session.commit()
flash("Network deleted")
return redirect(url_for("manage.list_networks"))
@bp.route("/peers", methods=["GET", "POST"])
@login_required
def list_peers():
peers = current_user.peers
return render_template("peer_list.html", peers=peers)
"""
Here starts the enrollment API
Maybe this should be in a new blueprint but i don't care enough about where
exactly it goes that it's fine.
"""
@bp.route("/enroll_start", methods=["GET"])
@login_required
def enroll_start():
"""
this function only creates a new enrollment request and gives the ID back to
the user for use in the client, which should proceed immediately to the next
step -- using the ID to complete the enrollment process by sending a public
key back, and receiving an IP in return. This should probably return a form
so users can choose which network (of the ones they manage) the device
should be in.
Alternately to that, there could just be a menu in the client that lets the
user do that themselves without going to the web client except to
authenticate.
Theoretically, there should be an optional approval mechanism where someone
with the is_admin flag can choose to get an email every time someone tries
to enroll after the request is completed, where the peer will exist and be
"enrolled" but no one will actually get the configuration until it's marked
as "approved" in the database
"""
en_req = EnrollRequest(
id=str(ulid.ulid()),
user=str(current_user.id),
expires=datetime.now() + timedelta(days=30)
)
db.session.add(en_req)
db.session.commit()
return render_template("enroll_id.html", id=en_req.id)
@bp.route("/enroll_end", methods=["POST"])
def finish_enroll():
"""
The thought here is that the client would POST the enroll key in the last
route and a public key, and this creates the Peer struct before returning it
as JSON for the client to use
We should take in a network ID argument so we know what CIDR to put this IP
address in.
"""
json = request.get_json()
network = Network.query.filter_by(id=str(json['network_id'])).first()
if network is None:
abort(404)
network = ipaddress.IPv4Network(network.subnet)
en_req = EnrollRequest.query.filter_by(id=str(json['enroll_id'])).first()
if en_req is None:
abort(404)
peers = db.session.execute(db.select(Peer)).scalars()
in_network = filter(lambda p:
ipaddress.IPv4Network(p.addr).subnet_of(network), peers)
for ip in network.hosts():
if ip in in_network:
continue
else:
# This is an IP that we can use!
peer = Peer(id=str(ulid.ulid()),addr=str(ip), description=json['hostname'], public_key=json['public_key'], owner_id=en_req.user)
db.session.add(peer)
db.session.commit()
db.session.delete(en_req)
return {"ip": peer.addr, "id": peer.id}
abort(400)
|
