from flask import Flask, request, redirect, url_for, render_template_string, flash, Response import pymysql from pymysql.cursors import DictCursor import os import ipaddress import json app = Flask(__name__) app.secret_key = os.environ.get("SECRET_KEY", "zmenit-na-vlastni-tajne-heslo") DB_HOST = os.environ.get("DB_HOST", "localhost") DB_PORT = int(os.environ.get("DB_PORT", "3306")) DB_USER = os.environ.get("DB_USER", "ipam_user") DB_PASSWORD = os.environ.get("DB_PASSWORD", "") DB_NAME = os.environ.get("DB_NAME", "ipam") def get_db_connection(): return pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, cursorclass=DictCursor, autocommit=True, charset="utf8mb4" ) def validate_ipv4(ip_text: str) -> bool: try: ipaddress.IPv4Address(ip_text) return True except Exception: return False INDEX_TEMPLATE = """ IPplan
{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %} {% if mode == 'list' %}

IPplan

Řazeno podle IP adresy
Export JSON + Přidat IP
{% if query %} Zrušit filtr {% endif %}
{% if query %} Nalezeno záznamů: {{ rows|length }} pro výraz {{ query }} {% else %} Celkem záznamů: {{ rows|length }} {% endif %}
{% if rows %} {% for row in rows %} {% endfor %}
IP adresa FQDN Účel Akce
{{ row.ipaddress }}
{{ row.fqdn or '' }} {{ row.purpose or '' }} Upravit
{% else %}
Žádné záznamy nebyly nalezeny.
{% endif %} {% elif mode == 'edit' %}

Upravit záznam

Zpět
{% elif mode == 'add' %}

Přidat záznam

Zpět
{% endif %}
""" @app.route("/") def index(): query = request.args.get("q", "").strip() conn = get_db_connection() try: with conn.cursor() as cur: if query: like_value = f"%{query}%" cur.execute(""" SELECT id, ipaddress, fqdn, purpose FROM iplist WHERE ipaddress LIKE %s OR fqdn LIKE %s OR purpose LIKE %s ORDER BY INET_ATON(ipaddress) ASC """, (like_value, like_value, like_value)) else: cur.execute(""" SELECT id, ipaddress, fqdn, purpose FROM iplist ORDER BY INET_ATON(ipaddress) ASC """) rows = cur.fetchall() finally: conn.close() return render_template_string( INDEX_TEMPLATE, mode="list", rows=rows, query=query ) @app.route("/export/json") def export_json(): query = request.args.get("q", "").strip() conn = get_db_connection() try: with conn.cursor() as cur: if query: like_value = f"%{query}%" cur.execute(""" SELECT ipaddress, fqdn, purpose FROM iplist WHERE ipaddress LIKE %s OR fqdn LIKE %s OR purpose LIKE %s ORDER BY INET_ATON(ipaddress) ASC """, (like_value, like_value, like_value)) else: cur.execute(""" SELECT ipaddress, fqdn, purpose FROM iplist ORDER BY INET_ATON(ipaddress) ASC """) rows = cur.fetchall() finally: conn.close() json_data = json.dumps(rows, ensure_ascii=False, indent=2) return Response( json_data, mimetype="application/json", headers={ "Content-Disposition": "attachment; filename=iplist.json" } ) @app.route("/add", methods=["GET", "POST"]) def add_item(): form_data = { "ipaddress": "", "fqdn": "", "purpose": "" } if request.method == "POST": form_data["ipaddress"] = request.form.get("ipaddress", "").strip() form_data["fqdn"] = request.form.get("fqdn", "").strip() form_data["purpose"] = request.form.get("purpose", "").strip() if not validate_ipv4(form_data["ipaddress"]): flash("Neplatná IPv4 adresa.", "error") return render_template_string(INDEX_TEMPLATE, mode="add", form_data=form_data) conn = get_db_connection() try: with conn.cursor() as cur: try: cur.execute(""" INSERT INTO iplist (ipaddress, fqdn, purpose) VALUES (%s, %s, %s) """, ( form_data["ipaddress"], form_data["fqdn"] or None, form_data["purpose"] or None )) flash("Záznam byl přidán.", "success") return redirect(url_for("index")) except pymysql.err.IntegrityError: flash("Tahle IP adresa už v databázi existuje.", "error") finally: conn.close() return render_template_string(INDEX_TEMPLATE, mode="add", form_data=form_data) @app.route("/edit/", methods=["GET", "POST"]) def edit_item(item_id): conn = get_db_connection() try: with conn.cursor() as cur: if request.method == "POST": ipaddress_value = request.form.get("ipaddress", "").strip() fqdn_value = request.form.get("fqdn", "").strip() purpose_value = request.form.get("purpose", "").strip() if not validate_ipv4(ipaddress_value): flash("Neplatná IPv4 adresa.", "error") else: try: cur.execute(""" UPDATE iplist SET ipaddress = %s, fqdn = %s, purpose = %s WHERE id = %s """, ( ipaddress_value, fqdn_value or None, purpose_value or None, item_id )) flash("Záznam byl uložen.", "success") return redirect(url_for("index")) except pymysql.err.IntegrityError: flash("Tahle IP adresa už v databázi existuje.", "error") cur.execute(""" SELECT id, ipaddress, fqdn, purpose FROM iplist WHERE id = %s """, (item_id,)) item = cur.fetchone() if not item: flash("Záznam nebyl nalezen.", "error") return redirect(url_for("index")) finally: conn.close() return render_template_string(INDEX_TEMPLATE, mode="edit", item=item) @app.route("/delete/", methods=["POST"]) def delete_item(item_id): conn = get_db_connection() try: with conn.cursor() as cur: cur.execute("SELECT ipaddress FROM iplist WHERE id = %s", (item_id,)) item = cur.fetchone() if not item: flash("Záznam nebyl nalezen.", "error") return redirect(url_for("index")) cur.execute("DELETE FROM iplist WHERE id = %s", (item_id,)) flash(f"Záznam {item['ipaddress']} byl smazán.", "success") finally: conn.close() return redirect(url_for("index")) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=False)