286 lines
14 KiB
Python
286 lines
14 KiB
Python
"""
|
||
Beheer › IP-blokkades. Geldt voor alle scholen.
|
||
|
||
- Administrators zien en beheren alles (blokkeren, altijd toestaan, schakelaar, testen).
|
||
- Een gewone gebruiker met het recht 'ip_whitelist' ziet enkel de toegestane adressen van de actieve school,
|
||
kan een adres toevoegen (school = actieve school, opmerking = zijn naam) en enkel zijn eigen regels verwijderen.
|
||
"""
|
||
import functools
|
||
|
||
from flask import Blueprint, render_template, request, flash, redirect, url_for, session
|
||
|
||
from blueprints.auth import login_required
|
||
from database import central_query
|
||
from rechten import admin_vereist, recht_vereist, is_admin, log
|
||
from services import ipblokkade as ipb
|
||
from tenancy import alle_scholen, school_op_code, huidige_school
|
||
|
||
ipblokkade_bp = Blueprint('ipblokkade', __name__)
|
||
|
||
|
||
def _schoolnaam(code):
|
||
s = school_op_code(code, ook_inactief=True)
|
||
return s.naam if s else code
|
||
|
||
|
||
def admin_of_whitelist(view):
|
||
"""Administrators altijd (ook zonder gekozen school); anderen met het recht 'ip_whitelist' in de actieve school."""
|
||
beschermd = recht_vereist('ip_whitelist')(view)
|
||
|
||
@functools.wraps(view)
|
||
def wrapped(*a, **kw):
|
||
return view(*a, **kw) if session.get('auth') and is_admin() else beschermd(*a, **kw)
|
||
return wrapped
|
||
|
||
|
||
# Gewone gebruikers mogen hoogstens zo'n reeks toestaan (een ruimere reeks laat te veel adressen altijd binnen)
|
||
MAX_PREFIX_GEBRUIKER = {4: 24, 6: 64}
|
||
|
||
|
||
def _eigen(r):
|
||
"""Mag de ingelogde (gewone) gebruiker deze regel verwijderen? Enkel als hij hem zelf toevoegde."""
|
||
gid = session.get('gebruiker_id')
|
||
return bool(gid) and r.get('Gebruiker_ID') == gid and r.get('Soort') == 'toestaan'
|
||
|
||
|
||
def _gekozen_scholen():
|
||
"""De aangevinkte scholen uit het formulier (enkel bestaande schoolcodes)."""
|
||
return ipb.scholen_waarde(c for c in request.form.getlist('scholen') if school_op_code(c, ook_inactief=True))
|
||
|
||
|
||
@ipblokkade_bp.route("/", methods=["GET"])
|
||
@login_required
|
||
@admin_of_whitelist
|
||
def overzicht():
|
||
ipb.ensure_table()
|
||
rijen = central_query("SELECT * FROM IP_blokkades ORDER BY Soort, Reeks") or []
|
||
if not is_admin():
|
||
return _overzicht_gebruiker(rijen)
|
||
import ipaddress
|
||
for r in rijen: # sorteersleutel: numeriek op adres (zodat 9.x vóór 85.x komt), dan op grootte van de reeks
|
||
try:
|
||
net = ipaddress.ip_network(r['Reeks'], strict=False)
|
||
r['Sorteer'] = f"{net.version}-{int(net.network_address):040d}-{net.prefixlen:03d}"
|
||
except ValueError:
|
||
r['Sorteer'] = r['Reeks']
|
||
r['Scholen_codes'] = ipb.scholen_codes(r.get('Scholen'))
|
||
r['Scholen_namen'] = sorted((_schoolnaam(c) for c in r['Scholen_codes']), key=str.lower)
|
||
test_ip = (request.args.get('test') or '').strip()
|
||
test = None
|
||
if test_ip:
|
||
try:
|
||
lijst = ipb.regels(vers=True)
|
||
geweigerd, rid = ipb.beoordeel(test_ip, lijst)
|
||
regel = next((r for r in rijen if r['ID'] == rid), None)
|
||
test = {'ip': test_ip, 'geweigerd': geweigerd, 'regel': regel}
|
||
except Exception as e:
|
||
flash(f"Test mislukt: {e}", "error")
|
||
modus = ipb.instelling(ipb.MODUS_SLEUTEL)
|
||
teller = ipb.instelling(ipb.TELLER_SLEUTEL)
|
||
scholen = sorted(alle_scholen(ook_inactief=True), key=lambda s: s.naam.lower())
|
||
return render_template("ipblokkade.html", rijen=rijen, mijn_ip=request.remote_addr, test=test, scholen=scholen,
|
||
blokkeren=[r for r in rijen if r['Soort'] == 'blokkeren'],
|
||
toestaan=[r for r in rijen if r['Soort'] == 'toestaan'],
|
||
enkel_toegestaan=bool(modus and modus['Waarde'] == '1'), modus=modus,
|
||
lokaal=ipb.is_lokaal(request.remote_addr),
|
||
modus_geweigerd=int(teller['Waarde']) if teller and str(teller['Waarde']).isdigit() else 0)
|
||
|
||
|
||
def _overzicht_gebruiker(rijen):
|
||
"""Gewone gebruiker: enkel de toegestane adressen van de actieve school."""
|
||
school = huidige_school()
|
||
eigen_school = [r for r in rijen if r['Soort'] == 'toestaan' and school.code in ipb.scholen_codes(r.get('Scholen'))]
|
||
import ipaddress
|
||
for r in eigen_school:
|
||
try:
|
||
net = ipaddress.ip_network(r['Reeks'], strict=False)
|
||
r['Sorteer'] = f"{net.version}-{int(net.network_address):040d}-{net.prefixlen:03d}"
|
||
except ValueError:
|
||
r['Sorteer'] = r['Reeks']
|
||
r['Eigen'] = _eigen(r)
|
||
return render_template("ipblokkade_school.html", rijen=eigen_school, school_naam=school.naam,
|
||
mijn_ip=request.remote_addr, lokaal=ipb.is_lokaal(request.remote_addr),
|
||
naam=session.get('naam') or session.get('user'))
|
||
|
||
|
||
@ipblokkade_bp.route("/modus", methods=["POST"])
|
||
@login_required
|
||
@admin_vereist
|
||
def modus():
|
||
"""Schakelaar: enkel nog adressen uit "Altijd toestaan" binnenlaten."""
|
||
ipb.ensure_table()
|
||
aan = request.form.get('aan') == '1'
|
||
if aan:
|
||
mijn = request.remote_addr
|
||
if ipb.is_lokaal(mijn):
|
||
flash(f"PIVOT ziet jouw adres als {mijn}. Dan draait PIVOT achter een reverse proxy (of test je lokaal) en "
|
||
"lijkt elke bezoeker van dat adres te komen: de witte lijst zou dan niets tegenhouden of iedereen "
|
||
"buitensluiten. Zet eerst in config.ini › [algemeen] proxy_aantal = 1 en herstart PIVOT; bij "
|
||
"'Jouw IP-adres' moet je echte adres verschijnen.", "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
lijst = ipb.regels(vers=True)
|
||
if not any(sz == 'toestaan' for _, sz, _ in lijst):
|
||
flash("Voeg eerst minstens één adres of reeks toe bij 'Altijd toestaan'.", "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
if ipb.beoordeel(mijn, lijst, enkel_toe=True)[0]:
|
||
flash(f"Jouw eigen adres ({mijn}) staat niet bij 'Altijd toestaan': dan sluit je jezelf buiten. "
|
||
"Voeg het eerst toe.", "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
ipb.zet_enkel_toegestaan(aan, session.get('user'))
|
||
flash("Enkel nog adressen uit 'Altijd toestaan' worden binnengelaten." if aan
|
||
else "Iedereen wordt weer binnengelaten, behalve de geblokkeerde adressen.", "success" if not aan else "warning")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
|
||
|
||
@ipblokkade_bp.route("/toevoegen", methods=["POST"])
|
||
@login_required
|
||
@admin_of_whitelist
|
||
def toevoegen():
|
||
if not is_admin():
|
||
return _toevoegen_gebruiker()
|
||
soort = 'toestaan' if request.form.get('soort') == 'toestaan' else 'blokkeren'
|
||
try:
|
||
net = ipb.normaliseer(request.form.get('reeks'))
|
||
except ipb.OngeldigeReeks as e:
|
||
flash(str(e), "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
ipb.ensure_table()
|
||
if soort == 'blokkeren':
|
||
# Niet jezelf buitensluiten: weigeren als je eigen adres erin valt (tenzij het uitdrukkelijk toegestaan is)
|
||
lijst = ipb.regels(vers=True) + [(0, 'blokkeren', net)]
|
||
if ipb.beoordeel(request.remote_addr, lijst)[0]:
|
||
flash(f"Deze reeks bevat je eigen IP-adres ({request.remote_addr}): dan sluit je jezelf buiten. "
|
||
"Voeg eerst je eigen adres toe als uitzondering (toestaan), of kies een kleinere reeks.", "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
bestaat = central_query("SELECT ID FROM IP_blokkades WHERE Reeks = %s AND Soort = %s", (str(net), soort))
|
||
if bestaat:
|
||
flash(f"{net} staat al in de lijst.", "warning")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
# Bij "Altijd toestaan" kan je aanduiden bij welke school/scholen het adres hoort (enkel ter info in de tabel)
|
||
scholen = _gekozen_scholen() if soort == 'toestaan' else None
|
||
central_query("INSERT INTO IP_blokkades (Reeks, Soort, Opmerking, Scholen, Aangemaakt, Door, Gebruiker_ID) "
|
||
"VALUES (%s, %s, %s, %s, NOW(), %s, %s)",
|
||
(str(net), soort, (request.form.get('opmerking') or '').strip()[:255] or None,
|
||
scholen, session.get('user'), session.get('gebruiker_id')),
|
||
commit=True)
|
||
ipb.vergeet_cache()
|
||
aantal = net.num_addresses
|
||
getal = f"{aantal:,}".replace(',', '.')
|
||
flash(f"{net} toegevoegd ({'geblokkeerd' if soort == 'blokkeren' else 'altijd toegestaan'}"
|
||
f"{', ' + ', '.join(_schoolnaam(c) for c in ipb.scholen_codes(scholen)) if scholen else ''}, {getal} adres{'sen' if aantal != 1 else ''}).", "success")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
|
||
|
||
def _toevoegen_gebruiker():
|
||
"""Gewone gebruiker: enkel 'Altijd toestaan', school = actieve school, opmerking = zijn naam."""
|
||
terug = redirect(url_for('ipblokkade.overzicht'))
|
||
try:
|
||
net = ipb.normaliseer(request.form.get('reeks'))
|
||
except ipb.OngeldigeReeks as e:
|
||
flash(str(e), "error")
|
||
return terug
|
||
grens = MAX_PREFIX_GEBRUIKER[net.version]
|
||
if net.prefixlen < grens:
|
||
flash(f"{net} is een te grote reeks. Je kan één adres toevoegen of hoogstens een /{grens}-reeks "
|
||
f"(bv. {'85.12.3.' if net.version == 4 else 'een /64'}). Vraag een administrator voor een grotere reeks.", "error")
|
||
return terug
|
||
ipb.ensure_table()
|
||
school = huidige_school()
|
||
naam = (session.get('naam') or session.get('user') or '').strip()[:255] or None
|
||
bestaat = central_query("SELECT ID, Scholen FROM IP_blokkades WHERE Reeks = %s AND Soort = 'toestaan'", (str(net),))
|
||
if bestaat:
|
||
codes = ipb.scholen_codes(bestaat[0]['Scholen'])
|
||
if school.code in codes:
|
||
flash(f"{net} staat al in de lijst van {school.naam}.", "warning")
|
||
else:
|
||
# Al toegestaan voor een andere school: deze school erbij zetten (de regel blijft van wie hem maakte)
|
||
central_query("UPDATE IP_blokkades SET Scholen = %s WHERE ID = %s",
|
||
(ipb.scholen_waarde(codes + [school.code]), bestaat[0]['ID']), commit=True)
|
||
flash(f"{net} was al toegestaan voor een andere school en staat nu ook bij {school.naam}.", "success")
|
||
return terug
|
||
central_query("INSERT INTO IP_blokkades (Reeks, Soort, Opmerking, Scholen, Aangemaakt, Door, Gebruiker_ID) "
|
||
"VALUES (%s, 'toestaan', %s, %s, NOW(), %s, %s)",
|
||
(str(net), naam, school.code, session.get('user'), session.get('gebruiker_id')), commit=True)
|
||
ipb.vergeet_cache()
|
||
log(f"IP toegestaan voor {school.code}: {net}")
|
||
flash(f"{net} toegevoegd aan de toegestane adressen van {school.naam}.", "success")
|
||
return terug
|
||
|
||
|
||
@ipblokkade_bp.route("/leegmaken", methods=["POST"])
|
||
@login_required
|
||
@admin_vereist
|
||
def leegmaken():
|
||
"""Verwijdert alle 'blokkeren'-regels. 'Altijd toestaan' en de schakelaar blijven ongemoeid."""
|
||
ipb.ensure_table()
|
||
rij = central_query("SELECT COUNT(*) AS n FROM IP_blokkades WHERE Soort = 'blokkeren'")
|
||
aantal = int(rij[0]['n']) if rij else 0
|
||
if not aantal:
|
||
flash("De bloklijst is al leeg.", "warning")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
central_query("DELETE FROM IP_blokkades WHERE Soort = 'blokkeren'", commit=True)
|
||
ipb.vergeet_cache()
|
||
try:
|
||
from rechten import log
|
||
log(f"Bloklijst IP-blokkades leeggemaakt ({aantal} regels)")
|
||
except Exception:
|
||
pass
|
||
flash(f"Bloklijst leeggemaakt: {aantal} regel{'s' if aantal != 1 else ''} verwijderd.", "success")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
|
||
|
||
@ipblokkade_bp.route("/<int:regel_id>/bewerken", methods=["POST"])
|
||
@login_required
|
||
@admin_vereist
|
||
def bewerken(regel_id):
|
||
"""Reeks, scholen en opmerking van een bestaande regel aanpassen (de soort blijft dezelfde)."""
|
||
terug = redirect(url_for('ipblokkade.overzicht'))
|
||
ipb.ensure_table()
|
||
rij = central_query("SELECT ID, Soort FROM IP_blokkades WHERE ID = %s", (regel_id,))
|
||
if not rij:
|
||
flash("Deze regel bestaat niet meer.", "error")
|
||
return terug
|
||
soort = rij[0]['Soort']
|
||
try:
|
||
net = ipb.normaliseer(request.form.get('reeks'))
|
||
except ipb.OngeldigeReeks as e:
|
||
flash(str(e), "error")
|
||
return terug
|
||
if central_query("SELECT ID FROM IP_blokkades WHERE Reeks = %s AND Soort = %s AND ID <> %s", (str(net), soort, regel_id)):
|
||
flash(f"{net} staat al in de lijst.", "warning")
|
||
return terug
|
||
# Niet jezelf buitensluiten: met de aangepaste regel moet je eigen adres nog binnen mogen
|
||
mijn = request.remote_addr
|
||
nieuw = [r for r in ipb.regels(vers=True) if r[0] != regel_id] + [(regel_id, soort, net)]
|
||
if ipb.beoordeel(mijn, nieuw, enkel_toe=ipb.enkel_toegestaan())[0] and not ipb.beoordeel(mijn, ipb.regels())[0]:
|
||
flash(f"Met deze aanpassing zou je eigen IP-adres ({mijn}) geweigerd worden: dan sluit je jezelf buiten.", "error")
|
||
return terug
|
||
central_query("UPDATE IP_blokkades SET Reeks = %s, Opmerking = %s, Scholen = %s WHERE ID = %s",
|
||
(str(net), (request.form.get('opmerking') or '').strip()[:255] or None,
|
||
_gekozen_scholen() if soort == 'toestaan' else None, regel_id), commit=True)
|
||
ipb.vergeet_cache()
|
||
flash(f"{net} bijgewerkt.", "success")
|
||
return terug
|
||
|
||
|
||
@ipblokkade_bp.route("/<int:regel_id>/verwijderen", methods=["POST"])
|
||
@login_required
|
||
@admin_of_whitelist
|
||
def verwijderen(regel_id):
|
||
if not is_admin():
|
||
rij = central_query("SELECT ID, Soort, Gebruiker_ID FROM IP_blokkades WHERE ID = %s", (regel_id,))
|
||
if not rij or not _eigen(rij[0]):
|
||
flash("Je kan enkel adressen verwijderen die je zelf toegevoegd hebt.", "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
# Staat de witte-lijstmodus aan, dan mag je de regel die jou zelf binnenlaat niet verwijderen
|
||
if ipb.enkel_toegestaan(vers=True):
|
||
rest = [r for r in ipb.regels(vers=True) if r[0] != regel_id]
|
||
if ipb.beoordeel(request.remote_addr, rest, enkel_toe=True)[0]:
|
||
flash("Deze regel laat jou zelf binnen terwijl enkel 'Altijd toestaan' geldt: verwijderen zou je "
|
||
"buitensluiten. Zet eerst de schakelaar uit of voeg je adres apart toe.", "error")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|
||
central_query("DELETE FROM IP_blokkades WHERE ID = %s", (regel_id,), commit=True)
|
||
ipb.vergeet_cache()
|
||
flash("Regel verwijderd.", "success")
|
||
return redirect(url_for('ipblokkade.overzicht'))
|