Merge branch 'main' of https://github.com/ZIT-P22/IP-Atlas

.vscode/settings.json

@@ -7,5 +7,14 @@
         "test_*.py"
     ],
     "python.testing.pytestEnabled": false,
-    "python.testing.unittestEnabled": true
+    "python.testing.unittestEnabled": true,
+    "sqltools.useNodeRuntime": true,
+    "sqltools.connections": [
+        {
+            "previewLimit": 50,
+            "driver": "SQLite",
+            "database": "${workspaceFolder:IP-Atlas}/database/ip_atlas.db",
+            "name": "ip-atlas"
+        }
+    ]
 }

ip-atlas/crud.py

@@ -0,0 +1,53 @@
+
+# this Script contains the CRUD operations for the ip-atlas database
+# Author: Janneck Lehmann
+# Date: 2024-02-07
+from models import db, Host, Tag, Port, HostTag, PortFB
+from helper import *
+
+
+
+# function which converts the given data from db to the json format
+def convert_to_json_format(host):
+    host_data = {
+        "id": host.id,
+        "name": host.hostname,
+        "ip": host.ipv4,
+        "ipv6": host.ipv6,  
+        "portsFB": [portFB.portFB_number for portFB in host.portsFB],
+        "tags": [host_tag.tag.tag_name for host_tag in host.tags],
+    }
+    return host_data
+
+# function which reads the host from the database by the given id
+def get_host_by_id(id):
+    host = Host.query.filter_by(id=id).first()
+    return convert_to_json_format(host)
+
+#
+
+# function which reads all hosts from the database where the attribute deleted is set to false
+def read_all_hosts():
+    hosts = Host.query.filter_by(deleted=False).all()
+    return hosts
+
+# function which writes the given data to the database
+def write_to_db(db, data):
+    if db == "host":
+        host = Host(hostname=data["name"], ipv4=data["ipv4"], ipv6=data["ipv6"])
+        db.session.add(host)
+    elif db == "port":
+        port = Port(host_id=data["host_id"], port_number=data["port_number"])
+        db.session.add(port)
+    elif db == "tag":
+        tag = Tag(tag_name=data["tag_name"])
+        db.session.add(tag)
+    elif db == "host_tag":
+        host_tag = HostTag(host_id=data["host_id"], tag_id=data["tag_id"])
+        db.session.add(host_tag)
+    db.session.commit()
+
+
+# function which updates the given data in the database
+
+

ip-atlas/helper.py

@@ -2,6 +2,20 @@
 import os
 import platform
 import subprocess
+from models import Host, Port, Tag
+
+#! db functions
+
+# function which checks if the given tag exists in the database and returns the id if it exists
+def tag_exists(tag_name, method="bool"):
+    tag = Tag.query.filter_by(tag_name=tag_name).first()
+    if method == "id":
+        if tag:
+            return tag.id
+        else:
+            return None
+    else:
+        return tag is not None
 
 
 # checks if the json document is there

ip-atlas/models.py

@@ -12,8 +12,8 @@ class Host(db.Model):
     hostname = Column(String, unique=True, nullable=False, index=True)
     ipv4 = Column(String, nullable=False, index=True)
     ipv6 = Column(String, index=True)
-    cidr = Column(Integer)
     deleted = Column(Boolean, default=False)
+    portsFB = relationship("PortFB", back_populates="host")
     ports = relationship("Port", back_populates="host")
     tags = relationship("HostTag", back_populates="host")
 
@@ -53,9 +53,14 @@ class Port(db.Model):
     id = Column(Integer, primary_key=True, autoincrement=True)
     host_id = Column(Integer, ForeignKey("hosts.id"))
     port_number = Column(Integer, nullable=False)
-    deleted = Column(Boolean, default=False)
     host = relationship("Host", back_populates="ports")
-
+
+class PortFB(db.Model):
+    __tablename__ = "portsFB"
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    host_id = Column(Integer, ForeignKey("hosts.id"))
+    portFB_number = Column(Integer, nullable=False)
+    host = relationship("Host", back_populates="portsFB")
 
 class AuditLog(db.Model):
     __tablename__ = "audit_logs"

ip-atlas/routes/atlas.py

@@ -2,7 +2,7 @@
 from jinja2 import TemplateNotFound
 from helper import *
 from filter import *
-from models import Host, Port, Tag
+from crud import *
 
 bp_atlas = Blueprint("atlas", __name__)
 
@@ -16,17 +16,10 @@ def index():
 
 @bp_atlas.route("/ip/list")
 def list():
-    hosts = Host.query.all()
     data = {"hosts": []}
+    hosts = read_all_hosts()
     for host in hosts:
-        host_data = {
-            "id": host.id,
-            "name": host.hostname,
-            "ip": host.ipv4,  # Assuming ipv4 is the primary IP to display
-            "ports": [port.port_number for port in host.ports],
-            "tags": [host_tag.tag.tag_name for host_tag in host.tags],  # Adjusted line
-        }
-        data["hosts"].append(host_data)
+        data["hosts"].append(convert_to_json_format(host))
     return render_template("ip/list.html", data=data)
 
 

ip-atlas/templates/ip/list.html

@@ -154,7 +154,7 @@ <h1 class="text-lg font-semibold md:text-2xl">IP Adressen</h1>
                 <tr data-id="{{ host.id }}" class="transition duration-150 ease-in-out border-b hover:bg-gray-700 hover:bg-opacity-25">
                     <td class="hidden p-4 align-middle edit-field md:table-cell">{{ host.name }}</td>
                     <td class="p-4 font-medium align-middle edit-field">{{ host.ip }}</td>
-                    <td class="p-4 align-middle edit-field">{{ host.ports|join(", ") }}</td>
+                    <td class="p-4 align-middle edit-field">{{ host.portsFB|join(", ") }}</td>
                     <td class="p-4 align-middle edit-field">
                         {% if host.tags %}
                         {{ host.tags|join(", ") }}

ip-atlas/testdatenfiller.py

@@ -1,7 +1,7 @@
 from faker import Faker
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
-from models import db, Host, Tag, Port, HostTag
+from models import db, Host, Tag, Port, HostTag, PortFB
 import random
 from app import atlasapp
 
@@ -20,6 +20,7 @@ def generate_test_data(num_hosts=50):
         ipv4 = fake.ipv4_private(network=False, address_class=None)
         ipv6 = fake.ipv6(network=False)
         ports_numbers = [fake.random_int(min=1, max=65535) for _ in range(2)]
+        portsFB_numbers = [fake.random_int(min=1, max=65535) for _ in range(2)]
 
         # Create Host instance
         host = Host(hostname=hostname, ipv4=ipv4, ipv6=ipv6)
@@ -30,6 +31,11 @@ def generate_test_data(num_hosts=50):
         for port_number in ports_numbers:
             port = Port(host_id=host.id, port_number=port_number)
             session.add(port)
+
+        # Create Port instances for FB
+        for portFB_number in portsFB_numbers:
+            portFB = PortFB(host_id=host.id, portFB_number=portFB_number)
+            session.add(portFB)
 
         # Create and associate Tags
         tags = [fake.word(), fake.word(), "test"]