197 lines
8.1 KiB
Python
197 lines
8.1 KiB
Python
"""
|
|
IP-blokkades: aanvragen van bepaalde IP-adressen of reeksen weigeren, voor alle scholen.
|
|
|
|
Regels staan in de centrale tabel IP_blokkades:
|
|
- Soort 'blokkeren' → elke aanvraag van zo'n adres krijgt 403 (ook de loginpagina).
|
|
- Soort 'toestaan' → uitzondering: dit adres mag altijd, ook als het in een geblokkeerde reeks valt.
|
|
|
|
Een reeks kan je schrijven als:
|
|
85. → alles wat met 85. begint (85.0.0.0/8)
|
|
85.12. → 85.12.0.0/16
|
|
85.12.3. → 85.12.3.0/24
|
|
85.12.3.4 → één adres
|
|
85.0.0.0/8 → CIDR-notatie (ook IPv6, bv. 2a02:1810::/32)
|
|
De lijst wordt 30 seconden in het geheugen gehouden; een wijziging in PIVOT geldt meteen.
|
|
Lukt het lezen van de database niet, dan wordt niemand geweigerd (de site blijft bereikbaar).
|
|
"""
|
|
import ipaddress
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
CACHE_SECONDEN = 30
|
|
_cache = {'t': 0, 'regels': []}
|
|
_lock = threading.Lock()
|
|
|
|
|
|
class OngeldigeReeks(ValueError):
|
|
pass
|
|
|
|
|
|
def normaliseer(reeks):
|
|
"""Tekst van de gebruiker → ip_network. '85.' → 85.0.0.0/8, '85.12.3.4' → 85.12.3.4/32."""
|
|
t = str(reeks or '').strip().replace(' ', '')
|
|
if not t:
|
|
raise OngeldigeReeks("Vul een IP-adres of reeks in.")
|
|
if '/' not in t and ':' not in t:
|
|
delen = t.rstrip('.').split('.') if t.endswith('.') or t.count('.') < 3 else None
|
|
if delen is not None:
|
|
if not 1 <= len(delen) <= 3 or not all(d.isdigit() and 0 <= int(d) <= 255 for d in delen):
|
|
raise OngeldigeReeks(f"'{reeks}' is geen geldige reeks. Voorbeelden: 85. 85.12. 85.12.3.4 85.0.0.0/8")
|
|
t = '.'.join(delen + ['0'] * (4 - len(delen))) + f"/{8 * len(delen)}"
|
|
try:
|
|
return ipaddress.ip_network(t, strict=False)
|
|
except ValueError:
|
|
raise OngeldigeReeks(f"'{reeks}' is geen geldig IP-adres of geldige reeks. "
|
|
"Voorbeelden: 85. 85.12. 85.12.3.4 85.0.0.0/8")
|
|
|
|
|
|
def ensure_table():
|
|
from database import central_query
|
|
central_query("""
|
|
CREATE TABLE IF NOT EXISTS PIVOT_instellingen (
|
|
Sleutel VARCHAR(50) NOT NULL PRIMARY KEY,
|
|
Waarde VARCHAR(255) NULL,
|
|
Bijgewerkt DATETIME NULL,
|
|
Door VARCHAR(100) NULL
|
|
) ENGINE=InnoDB""", commit=True)
|
|
central_query("""
|
|
CREATE TABLE IF NOT EXISTS IP_blokkades (
|
|
ID INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
|
Reeks VARCHAR(64) NOT NULL,
|
|
Soort VARCHAR(10) NOT NULL DEFAULT 'blokkeren',
|
|
Opmerking VARCHAR(255) NULL,
|
|
Scholen VARCHAR(1000) NULL,
|
|
Aangemaakt DATETIME NOT NULL,
|
|
Door VARCHAR(100) NULL,
|
|
Gebruiker_ID INT NULL,
|
|
Aantal_geweigerd INT NOT NULL DEFAULT 0,
|
|
Laatst_geweigerd DATETIME NULL,
|
|
UNIQUE KEY uq_ip_reeks (Reeks, Soort)
|
|
) ENGINE=InnoDB""", commit=True)
|
|
# Bestaande tabel van vóór de schoolkeuze bij "Altijd toestaan" (of met de eerdere kolom School_code: één school)
|
|
kolommen = {r['Field'].lower() for r in central_query("SHOW COLUMNS FROM IP_blokkades") or []}
|
|
if 'scholen' not in kolommen:
|
|
central_query("ALTER TABLE IP_blokkades ADD COLUMN Scholen VARCHAR(1000) NULL AFTER Opmerking", commit=True)
|
|
if 'school_code' in kolommen:
|
|
central_query("UPDATE IP_blokkades SET Scholen = School_code WHERE School_code IS NOT NULL", commit=True)
|
|
if 'school_code' in kolommen:
|
|
central_query("ALTER TABLE IP_blokkades DROP COLUMN School_code", commit=True)
|
|
if 'gebruiker_id' not in kolommen: # wie de regel toevoegde (gewone gebruikers mogen enkel hun eigen regels verwijderen)
|
|
central_query("ALTER TABLE IP_blokkades ADD COLUMN Gebruiker_ID INT NULL AFTER Door", commit=True)
|
|
|
|
|
|
def scholen_codes(waarde):
|
|
"""'a,b' (kolom Scholen) → ['a', 'b']."""
|
|
return [c for c in (x.strip() for x in str(waarde or '').split(',')) if c]
|
|
|
|
|
|
def scholen_waarde(codes):
|
|
"""['a', 'b'] → 'a,b' (of None), zonder dubbels, in de gegeven volgorde."""
|
|
uit = []
|
|
for c in codes or []:
|
|
c = str(c).strip()
|
|
if c and c not in uit:
|
|
uit.append(c)
|
|
return ','.join(uit)[:1000] or None
|
|
|
|
|
|
MODUS_SLEUTEL = 'ip_enkel_toegestaan' # '1' = enkel adressen uit "Altijd toestaan" mogen binnen
|
|
TELLER_SLEUTEL = 'ip_whitelist_geweigerd' # aantal weigeringen door die modus
|
|
|
|
|
|
def _laad():
|
|
from database import central_query
|
|
uit = []
|
|
for r in central_query("SELECT ID, Reeks, Soort FROM IP_blokkades") or []:
|
|
try:
|
|
uit.append((r['ID'], r['Soort'], ipaddress.ip_network(r['Reeks'], strict=False)))
|
|
except ValueError:
|
|
continue
|
|
rij = central_query("SELECT Waarde FROM PIVOT_instellingen WHERE Sleutel = %s", (MODUS_SLEUTEL,))
|
|
_cache.update(t=time.time(), regels=uit, enkel_toegestaan=bool(rij and rij[0]['Waarde'] == '1'))
|
|
|
|
|
|
def regels(vers=False):
|
|
"""[(id, soort, ip_network)] uit de database (gecachet)."""
|
|
if vers or time.time() - _cache['t'] >= CACHE_SECONDEN:
|
|
with _lock:
|
|
if vers or time.time() - _cache['t'] >= CACHE_SECONDEN:
|
|
_laad()
|
|
return _cache['regels']
|
|
|
|
|
|
def enkel_toegestaan(vers=False):
|
|
"""True als de modus "enkel Altijd toestaan" aan staat."""
|
|
regels(vers)
|
|
return _cache.get('enkel_toegestaan', False)
|
|
|
|
|
|
def zet_enkel_toegestaan(aan, door=None):
|
|
from database import central_query
|
|
central_query("INSERT INTO PIVOT_instellingen (Sleutel, Waarde, Bijgewerkt, Door) VALUES (%s, %s, NOW(), %s) "
|
|
"ON DUPLICATE KEY UPDATE Waarde = VALUES(Waarde), Bijgewerkt = NOW(), Door = VALUES(Door)",
|
|
(MODUS_SLEUTEL, '1' if aan else '0', door), commit=True)
|
|
vergeet_cache()
|
|
|
|
|
|
def instelling(sleutel):
|
|
from database import central_query
|
|
rij = central_query("SELECT Waarde, Bijgewerkt, Door FROM PIVOT_instellingen WHERE Sleutel = %s", (sleutel,))
|
|
return rij[0] if rij else None
|
|
|
|
|
|
def is_lokaal(ip):
|
|
"""127.0.0.1 of ::1: dan ziet PIVOT waarschijnlijk een reverse proxy i.p.v. de echte bezoeker."""
|
|
try:
|
|
a = ipaddress.ip_address(str(ip or '').strip())
|
|
except ValueError:
|
|
return False
|
|
return a.is_loopback
|
|
|
|
|
|
def vergeet_cache():
|
|
_cache['t'] = 0
|
|
|
|
|
|
def beoordeel(ip, lijst=None, enkel_toe=None):
|
|
"""Geeft (geweigerd, regel_id). Een 'toestaan'-regel wint altijd van een 'blokkeren'-regel.
|
|
In de modus "enkel Altijd toestaan" wordt alles geweigerd wat niet op die lijst staat (regel_id = 'modus')."""
|
|
try:
|
|
adres = ipaddress.ip_address(str(ip or '').strip())
|
|
except ValueError:
|
|
return False, None
|
|
if getattr(adres, 'ipv4_mapped', None): # ::ffff:85.1.2.3 → 85.1.2.3
|
|
adres = adres.ipv4_mapped
|
|
lijst = regels() if lijst is None else lijst
|
|
if any(s == 'toestaan' and adres in net for _, s, net in lijst if net.version == adres.version):
|
|
return False, None
|
|
if enkel_toegestaan() if enkel_toe is None else enkel_toe:
|
|
return True, 'modus'
|
|
for rid, s, net in lijst:
|
|
if s == 'blokkeren' and net.version == adres.version and adres in net:
|
|
return True, rid
|
|
return False, None
|
|
|
|
|
|
def controleer(ip):
|
|
"""Voor elke aanvraag: True = weigeren. Telt de weigering bij de regel. Faalt nooit (fout → doorlaten)."""
|
|
try:
|
|
geweigerd, rid = beoordeel(ip)
|
|
except Exception as e:
|
|
sys.stderr.write(f"PIVOT: IP-blokkades lezen mislukt: {e}\n")
|
|
return False
|
|
if geweigerd:
|
|
try:
|
|
from database import central_query
|
|
if rid == 'modus':
|
|
central_query("INSERT INTO PIVOT_instellingen (Sleutel, Waarde, Bijgewerkt) VALUES (%s, '1', NOW()) "
|
|
"ON DUPLICATE KEY UPDATE Waarde = CAST(Waarde AS UNSIGNED) + 1, Bijgewerkt = NOW()",
|
|
(TELLER_SLEUTEL,), commit=True)
|
|
else:
|
|
central_query("UPDATE IP_blokkades SET Aantal_geweigerd = Aantal_geweigerd + 1, "
|
|
"Laatst_geweigerd = NOW() WHERE ID = %s", (rid,), commit=True)
|
|
except Exception:
|
|
pass
|
|
return geweigerd
|