centralizing parsing of nmap results

README.md

@@ -44,7 +44,7 @@ Swagger API is accessible from: [http://localhost:8000/docs](http://localhost:80
 
 ### Local Installation from Sources
 
-Clone the repository. Then, run `docker-compose up --build`.
+Clone the repository. Then, run `docker compose up --build`.
 
 ## Development
 

backend/api/info.py

@@ -2,7 +2,7 @@
 from fastapi import APIRouter
 import FindMyIP as ip
 from loguru import logger
-from scan.nmap import NmapScanner
+from service.nmap import NmapScanner
 
 router = APIRouter()
 

backend/api/profile.py

@@ -13,6 +13,7 @@ def get_profile(profile_id: str, db: Session = Depends(get_db)):
     """
     Get a profile by ID.
     """
+
     profile = db.query(models.Profile).filter(models.Profile.profile_id == profile_id).first()
     return profile
 

backend/api/scan.py

@@ -1,20 +1,22 @@
 from fastapi import APIRouter, Depends
 from backend.db import get_db
-from backend.schemas import ProfileRead
-from scan.nmap import NmapScanner
+from backend.schemas import ProfileOnlyRead
+from backend.service.profile_scan import ProfileScanService
+from service.nmap import NmapScanner
 from sqlalchemy.orm import Session
 
 router = APIRouter()
 
 
-@router.get("/scan/profile/{profile_id}", response_model=ProfileRead, tags=["scan"])
+@router.get("/scan/profile/{profile_id}", response_model=ProfileOnlyRead, tags=["scan"])
 async def scan_profile(profile_id: str, db: Session = Depends(get_db)):
     """
-    Scan the target profile .
+    Scan the target profile.
 
     """
-    nmap_scanner = NmapScanner()
-    profile = nmap_scanner.scan_profile(profile_id, db)
+    profile_service = ProfileScanService(profile_id, db)
+    profile = profile_service.scan_profile()
+
     return profile
 
 

backend/models.py

@@ -6,16 +6,8 @@
 from sqlalchemy.sql import func
 from sqlalchemy import String, ForeignKey
 from sqlalchemy.orm import relationship
-from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped
 from sqlalchemy.orm import mapped_column
-from sqlalchemy.orm import relationship
-from typing import List
-
-from sqlalchemy import ForeignKey
-from sqlalchemy.orm import Mapped
-from sqlalchemy.orm import mapped_column
-from sqlalchemy.orm import relationship
 
 
 class Profile(Base):
@@ -31,7 +23,7 @@ class Profile(Base):
     last_scan: Mapped[DateTime] = mapped_column(DateTime, nullable=True)
     created_at: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), nullable=False)
     updated_at: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), onupdate=func.now(), nullable=False)
-    scan_events: Mapped[List["ScanEvent"]] = relationship(back_populates="profile")
+    scan_events: Mapped[List["ScanEvent"]] = relationship(back_populates="profile", lazy="dynamic")
 
 
 class ScanEvent(Base):
@@ -40,43 +32,47 @@ class ScanEvent(Base):
     """
 
     __tablename__ = "scan"
-    scan_id: Mapped[str] = mapped_column(primary_key=True)
+    scan_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, nullable=False)
     profile_id: Mapped[str] = mapped_column(ForeignKey("profile.profile_id"))
+    scan_command: Mapped[str] = mapped_column(String, nullable=False)  # "@args": "/opt/homebrew/bin/nmap -sn -T4 -oX -
+    scan_start: Mapped[int] = mapped_column(Integer, nullable=False)
+    scan_end: Mapped[int] = mapped_column(Integer, nullable=False)
+    scan_status: Mapped[str] = mapped_column(String, nullable=False)
+    created_at: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), nullable=False)
     profile: Mapped[Profile] = relationship("Profile", back_populates="scan_events")
-    scan_command: Mapped[str] = mapped_column(String, nullable=True)  # "@args": "/opt/homebrew/bin/nmap -sn -T4 -oX -
-    scan_start: Mapped[int] = mapped_column(Integer, nullable=True)
-    scan_end: Mapped[int] = mapped_column(Integer, nullable=True)
-    scan_status: Mapped[str] = mapped_column(String, nullable=True)
+    hosts: Mapped[List["Host"]] = relationship(back_populates="scan")
 
 
-# class Host(Base):
-#     """
-#     A host is just a device on the network found during a scan event.
-#     """
-
-#     __tablename__ = "host"
-#     host_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
-#     scan_id = Column(UUID(as_uuid=True), nullable=False)
-#     start_time = Column(DateTime, nullable=False)
-#     end_time = Column(DateTime, nullable=False)
-#     state = Column(String, nullable=False)
-#     reason = Column(String, nullable=True)
-#     created_at = Column(DateTime, default=func.now(), nullable=False)
+class Host(Base):
+    """
+    A host is just a device on the network found during a scan event. One Scan can have many hosts.
+    """
 
+    __tablename__ = "host"
+    host_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    scan_id: Mapped[int] = mapped_column(ForeignKey("scan.scan_id"))
+    start_time: Mapped[int] = mapped_column(Integer, nullable=False)
+    end_time: Mapped[int] = mapped_column(Integer, nullable=False)
+    state: Mapped[str] = mapped_column(String, nullable=False)
+    reason: Mapped[str] = mapped_column(String, nullable=True)
+    created_at: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), nullable=False)
+    scan: Mapped[ScanEvent] = relationship("ScanEvent", back_populates="hosts")
+    addresses: Mapped[List["Address"]] = relationship(back_populates="host")
 
-# class Address(Base):
-#     """
-#     Address model for hosts. A Host can have many addresses.
-#     """
 
-#     __tablename__ = "address"
-#     address_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
-#     host_id = Column(UUID(as_uuid=True), nullable=False)
+class Address(Base):
+    """
+    Address model for hosts. A Host can have many addresses.
+    """
 
-#     address = Column(String, nullable=False)
-#     address_type = Column(String, nullable=True)
-#     vendor = Column(String, nullable=True)
-#     created_at = Column(DateTime, default=func.now(), nullable=False)
+    __tablename__ = "address"
+    address_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    host_id: Mapped[str] = mapped_column(ForeignKey("host.host_id"))
+    address: Mapped[str] = mapped_column(String, nullable=False)
+    address_type: Mapped[str] = mapped_column(String, nullable=False)
+    vendor: Mapped[str] = mapped_column(String, nullable=True)
+    created_at: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), nullable=False)
+    host: Mapped[Host] = relationship("Host", back_populates="addresses")
 
 
 # class HostName(Base):

backend/schemas.py

@@ -1,6 +1,20 @@
 from datetime import datetime
-from typing import List, Optional
+from typing import Optional
 from pydantic import BaseModel
+from typing import List, Optional
+
+
+class ProfileOnlyRead(BaseModel):
+    profile_id: str
+    profile_name: str
+    profile_description: Optional[str] = None
+    ip_range: str
+    last_scan: Optional[datetime] = None
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        from_attributes = True
 
 
 class ProfileRead(BaseModel):
@@ -11,6 +25,20 @@ class ProfileRead(BaseModel):
     last_scan: Optional[datetime] = None
     created_at: datetime
     updated_at: datetime
+    scan_events: List["ScanEventRead"] = []
+
+    class Config:
+        from_attributes = True
+
+
+class ScanEventRead(BaseModel):
+    scan_id: int
+    profile_id: str
+    scan_command: str
+    scan_start: datetime
+    scan_end: datetime
+    scan_status: str
+    created_at: datetime
 
     class Config:
         from_attributes = True
@@ -22,9 +50,3 @@ class ProfileBaseSchema(BaseModel):
     ip_range: str
     created_at: datetime | None = None
     updated_at: datetime | None = None
-
-
-class ListProfileResponse(BaseModel):
-    status: str
-    results: int
-    notes: List[ProfileBaseSchema]

backend/scan/nmap.py → backend/service/nmap.py

@@ -106,14 +106,16 @@ def get_scan_command(self):
                 case (
                     ScanTypesEnum.PING
                 ):  # No port scan. Yes traceroute sudo nmap -sn --traceroute -T4 -oX - -v 192.168.1.196
+                    # added 'unprivileged' flag to fix where nmap was showing all hosts as up, even though they weren't while running in docker.
                     flags = [
                         "-sn",
+                        "--unprivileged",
                         "-T4",
                         "-oX",
                         "-",
                     ]
                 case ScanTypesEnum.DETAILED:  # TCP SYN scan nmap -sS --min-rate 2000 -oX -
-                    flags = ["-sS", "-Pn", "--min-rate", "2000", "-oX", "-"]
+                    flags = ["-sS", "--min-rate", "2000", "-oX", "-"]
                 case ScanTypesEnum.OS:  # Enable OS detection only
                     flags = ["-sS", "-O", "--min-rate", "2000", "-oX", "-"]
                 case ScanTypesEnum.LIST:  # List scan sudo nmap -sL 192.168.1.200-210
@@ -135,7 +137,10 @@ def get_scan_command(self):
                 # To only show vulnerabilities within a certain range, add the following flag to the command where “x.x” is the CVSS score (ex: 6.5).
                 case _:
                     flags = []  # just use ping as the default.
-            command = [nmap_path] + flags + [self.target]
+            # command = [nmap_path] + flags + [self.target]
+            command = (
+                [nmap_path] + flags + self.target.split()
+            )  # this version works with a list of IPs concatenated with a space
             return command
         return None
 
@@ -227,7 +232,7 @@ def __scan(self) -> str:
         logger.info("Running command: {}".format(" ".join(command)))
         process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         try:
-            stdout, stderr = process.communicate(timeout=30)  # TODO: timeout should be configurable
+            stdout, stderr = process.communicate(timeout=300)  # TODO: timeout should be configurable
             # logger.info("Scan Results: {}".format(stdout.decode("utf-8")))
             json_stdout_response = json.dumps(xmltodict.parse(stdout.decode("utf-8")), indent=4)
 

backend/service/nmap_parser.py

@@ -0,0 +1,59 @@
+from typing import List
+
+from loguru import logger
+from backend.models import Address, Host, Profile, ScanEvent
+
+
+class NMapParserService(object):
+    """
+    An attempt at a service that can centralize nmap parsing, so ONE `ScanEvent` can be created from the scan results.
+    """
+
+    def __init__(self, scan_results: dict = None, profile: Profile = None):
+        self.scan_results = scan_results
+        self.profile = profile
+
+    def build_scan_event_from_results(self) -> ScanEvent:
+
+        scan_event = ScanEvent()
+        scan_event.profile_id = self.profile.profile_id
+        scan_event.profile = self.profile
+        scan_event.scan_command = self.scan_results["nmaprun"]["@args"]
+        scan_event.scan_start = self.scan_results["nmaprun"]["@start"]
+        scan_event.scan_end = self.scan_results["nmaprun"]["runstats"]["finished"]["@time"]
+        scan_event.scan_status = self.scan_results["nmaprun"]["runstats"]["finished"]["@exit"]
+
+        hosts: List[Host] = []
+        host_list = self.scan_results["nmaprun"]["host"]
+        for host in host_list:
+            try:
+                host_obj = Host()
+                host_obj.start_time = host.get("@starttime", None)
+                host_obj.end_time = host.get("@endtime", None)
+                host_status = host.get("status", "UNKNOWN")
+                if host_status:
+                    host_obj.state = host_status.get("@state", None)
+                    host_obj.reason = host_status.get("@reason", None)
+
+                address = host["address"]
+                if address:
+                    if isinstance(address, dict):
+                        host_address = Address()
+                        host_address.address_type = address.get("@addrtype", None)
+                        host_address.address = address["@addr"]
+                        host_address.vendor = address.get("@vendor", None)
+                        host_obj.addresses.append(host_address)
+                    elif isinstance(address, list):
+                        for addr in address:
+                            host_address = Address()
+                            host_address.address_type = addr.get("@addrtype", None)
+                            host_address.address = addr["@addr"]
+                            host_address.vendor = addr.get("@vendor", None)
+                            host_obj.addresses.append(host_address)
+                hosts.append(host_obj)
+            except Exception as e:
+                logger.error(f"Error parsing host for profile {self.profile.profile_id}: {e}")
+
+        scan_event.hosts = hosts
+
+        return scan_event

backend/service/profile_scan.py

@@ -0,0 +1,80 @@
+from datetime import datetime
+from loguru import logger
+from backend import models
+from sqlalchemy.orm import Session
+
+from backend.service.nmap import NmapScanner
+from backend.service.nmap_parser import NMapParserService
+
+
+class ProfileScanService(object):
+    """
+    This class is responsible for scanning a profile. It will leverage the nmap service for a multi-step scan and interact with the database to store the results.
+    """
+
+    def __init__(self, profile_id: str = None, db: Session = None):
+        self.profile_id = profile_id
+        self.db = db
+
+    def get_profile(self):
+        profile = self.db.query(models.Profile).filter(models.Profile.profile_id == self.profile_id).first()
+        if not profile:
+            logger.error("Profile not found.")
+            return None
+
+        self.profile = profile
+        return profile
+
+    def scan_profile(self):
+        """
+        This has to be done as a two-part scan. First we will scan using a ping scan to get the IP addresses of hosts that are "up"
+        and then we will scan the IP addresses using "detailed" mode.
+        """
+
+        profile = self.get_profile()
+        nmap_scanner = NmapScanner()
+
+        scan_results = {}
+        try:
+            ping_results = nmap_scanner.scan(
+                profile.ip_range, "ping"
+            )  # TODO: make this configurable, because not all networks will need a two-step scan
+            up_hosts = []
+            if ping_results:
+
+                nmap_hosts = ping_results["nmaprun"]["host"]
+                for host in nmap_hosts:
+                    if host["status"]["@state"] == "up":
+                        address = host["address"]  # address can be a dict or a list
+                        if isinstance(address, dict):
+                            if address["@addrtype"] == "ipv4":  # TODO: handle ipv6
+                                up_hosts.append(host["address"]["@addr"])
+                        elif isinstance(address, list):
+                            for addr in address:
+                                if addr["@addrtype"] == "ipv4":
+                                    up_hosts.append(addr["@addr"])
+
+                logger.info(f"Hosts that are up: {up_hosts}")
+                if len(up_hosts) > 0:
+                    concat_ips: str = " ".join(up_hosts)
+                    scan_results = nmap_scanner.scan(concat_ips, "detailed")
+                    if scan_results and "nmaprun" in scan_results:
+
+                        nmap_parser = NMapParserService(scan_results, profile)
+
+                        scan_event = nmap_parser.build_scan_event_from_results()
+                        profile.scan_events.append(scan_event)
+                        profile.last_scan = datetime.now()
+                        self.db.add(profile)
+                        self.db.commit()
+
+        except Exception as e:
+            logger.error(f"Error scanning profile {self.profile_id}: {e}")
+            return
+
+        # now let's scan the IP addresses that are "up"
+
+        return profile
+
+    def __process_scan_results(self, scan_results: dict) -> models.Profile:
+        pass

docker-compose.yml

@@ -1,5 +1,3 @@
-version: "3.7"
-
 services:
   backend:
     build: