From bcf628a17b0a7f75cb653a3296a11581f9e89949 Mon Sep 17 00:00:00 2001 From: hubaceks Date: Fri, 10 Apr 2026 22:42:48 +0200 Subject: [PATCH] first commit --- README.md | 0 app.py | 546 ++++++++++++++++++++++++++++++++++++++++++++++++++ ippam.service | 18 ++ 3 files changed, 564 insertions(+) create mode 100644 README.md create mode 100644 app.py create mode 100644 ippam.service diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/app.py b/app.py new file mode 100644 index 0000000..429ff74 --- /dev/null +++ b/app.py @@ -0,0 +1,546 @@ +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
+
+ +
+ +
+ + + {% 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 adresaFQDNÚčelAkce
+
+ {{ 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) \ No newline at end of file diff --git a/ippam.service b/ippam.service new file mode 100644 index 0000000..d0ed7ca --- /dev/null +++ b/ippam.service @@ -0,0 +1,18 @@ +[Unit] +Description=Simple IPAM web app +After=network.target + +[Service] +User=root +WorkingDirectory=/opt/ipplan +Environment="DB_HOST=localhost" +Environment="DB_PORT=3306" +Environment="DB_USER=ipam_user" +Environment="DB_PASSWORD=Megakrutoprisneheslo" +Environment="DB_NAME=ipam" +Environment="SECRET_KEY=nejake-dlouhe-tajne-heslo" +ExecStart=/usr/bin/python3 /opt/ipplan/app.py +Restart=always + +[Install] +WantedBy=multi-user.target \ No newline at end of file