#!/usr/bin/env python3

# ***********************IMPORTANT NMAP LICENSE TERMS************************
# *
# * The Nmap Security Scanner is (C) 1996-2024 Nmap Software LLC ("The Nmap
# * Project"). Nmap is also a registered trademark of the Nmap Project.
# *
# * This program is distributed under the terms of the Nmap Public Source
# * License (NPSL). The exact license text applying to a particular Nmap
# * release or source code control revision is contained in the LICENSE
# * file distributed with that version of Nmap or source code control
# * revision. More Nmap copyright/legal information is available from
# * https://nmap.org/book/man-legal.html, and further information on the
# * NPSL license itself can be found at https://nmap.org/npsl/ . This
# * header summarizes some key points from the Nmap license, but is no
# * substitute for the actual license text.
# *
# * Nmap is generally free for end users to download and use themselves,
# * including commercial use. It is available from https://nmap.org.
# *
# * The Nmap license generally prohibits companies from using and
# * redistributing Nmap in commercial products, but we sell a special Nmap
# * OEM Edition with a more permissive license and special features for
# * this purpose. See https://nmap.org/oem/
# *
# * If you have received a written Nmap license agreement or contract
# * stating terms other than these (such as an Nmap OEM license), you may
# * choose to use and redistribute Nmap under those terms instead.
# *
# * The official Nmap Windows builds include the Npcap software
# * (https://npcap.com) for packet capture and transmission. It is under
# * separate license terms which forbid redistribution without special
# * permission. So the official Nmap Windows builds may not be redistributed
# * without special permission (such as an Nmap OEM license).
# *
# * Source is provided to this software because we believe users have a
# * right to know exactly what a program is going to do before they run it.
# * This also allows you to audit the software for security holes.
# *
# * Source code also allows you to port Nmap to new platforms, fix bugs, and
# * add new features. You are highly encouraged to submit your changes as a
# * Github PR or by email to the dev@nmap.org mailing list for possible
# * incorporation into the main distribution. Unless you specify otherwise, it
# * is understood that you are offering us very broad rights to use your
# * submissions as described in the Nmap Public Source License Contributor
# * Agreement. This is important because we fund the project by selling licenses
# * with various terms, and also because the inability to relicense code has
# * caused devastating problems for other Free Software projects (such as KDE
# * and NASM).
# *
# * The free version of Nmap is distributed in the hope that it will be
# * useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. Warranties,
# * indemnification and commercial support are all available through the
# * Npcap OEM program--see https://nmap.org/oem/
# *
# ***************************************************************************/

import os
import os.path
import re
import io
import unittest

from glob import glob

from zenmapCore.Name import APP_NAME
from zenmapCore.NmapOptions import NmapOptions
from zenmapCore.NmapParser import NmapParser
from zenmapCore.UmitLogging import log


class HostSearch(object):
    @staticmethod
    def match_target(host, name):
        name = name.lower()
        mac = host.get_mac()
        ip = host.get_ip()
        ipv6 = host.get_ipv6()

        if mac and 'addr' in mac:
            if name in mac['addr'].lower():
                return True
        if ip and 'addr' in ip:
            if name in ip['addr'].lower():
                return True
        if ipv6 and 'addr' in ipv6:
            if name in ipv6['addr'].lower():
                return True

        if HostSearch.match_hostname(host, name):
            return True
        return False

    @staticmethod
    def match_hostname(host, hostname):
        hostname = hostname.lower()
        hostnames = host.get_hostnames()
        for hn in hostnames:
            if hostname in hn['hostname'].lower():
                return True
        else:
            return False

    @staticmethod
    def match_service(host, service):
        for port in host.get_ports():
            # We concatenate all useful fields and add them to the list
            if port['port_state'] not in ['open', 'open|filtered']:
                continue
            version = " ".join(
                    port.get(x, "") for x in (
                        "service_name",
                        "service_product",
                        "service_version",
                        "service_extrainfo"
                        )
                    )

            if service in version.lower():
                return True
        else:
            return False

    @staticmethod
    def match_os(host, os):
        os = os.lower()

        osmatches = host.get_osmatches()

        for osmatch in osmatches:
            os_str = osmatch['name'].lower()
            for osclass in osmatch['osclasses']:
                os_str += " " + osclass['vendor'].lower() + " " +\
                          osclass['osfamily'].lower() + " " +\
                          osclass['type'].lower()
            if os in os_str:
                return True

        return False

    @staticmethod
    def match_port(host_ports, port, port_state):
        # Check if the port is parsable, if not return False silently
        if re.match(r"^\d+$", port) is None:
            return False

        for hp in host_ports:
            if hp['portid'] == port and hp['port_state'] == port_state:
                return True

        return False


class SearchResult(object):
    def __init__(self):
        """This constructor is always called by SearchResult subclasses."""
        pass

    def search(self, **kargs):
        """Performs a search on each parsed scan. Since the 'and' operator is
        implicit, the search fails as soon as one of the tests fails. The
        kargs argument is a map having operators as keys and argument lists as
        values."""

        for scan_result in self.get_scan_results():
            self.parsed_scan = scan_result

            # Test each given operator against the current parsed result
            for operator, args in kargs.items():
                if not self._match_all_args(operator, args):
                    # No match => we discard this scan_result
                    break
            else:
                # All operator-matching functions have returned True, so this
                # scan_result satisfies all conditions
                yield self.parsed_scan

    def _match_all_args(self, operator, args):
        """A helper function that calls the matching function for the given
        operator and each of its arguments."""
        for arg in args:
            positive = True
            if arg != "" and arg[0] == "!":
                arg = arg[1:]
                positive = False
            if positive != self.__getattribute__("match_%s" % operator)(arg):
                # No match for this operator
                return False
        else:
            # All arguments for this operator produced a match
            return True

    def get_scan_results(self):
        # To be implemented by classes that are going to inherit this one
        pass

    def basic_match(self, keyword, property):
        if keyword == "*" or keyword == "":
            return True

        return keyword.lower() in str(
                self.parsed_scan.__getattribute__(property)).lower()

    def match_keyword(self, keyword):
        log.debug("Match keyword: %s" % keyword)

        return self.basic_match(keyword, "nmap_output") or \
               self.match_profile(keyword) or \
               self.match_target(keyword)

    def match_profile(self, profile):
        log.debug("Match profile: %s" % profile)
        log.debug("Comparing: %s == %s ??" % (
            str(self.parsed_scan.profile_name).lower(),
            "*%s*" % profile.lower()))
        return (profile == "*" or profile == "" or
                profile.lower() in str(self.parsed_scan.profile_name).lower())

    def match_option(self, option):
        log.debug("Match option: %s" % option)

        if option == "*" or option == "":
            return True

        ops = NmapOptions()
        ops.parse_string(self.parsed_scan.get_nmap_command())

        if "(" in option and ")" in option:
            # The syntax allows matching option arguments as
            # "opt:option_name(value)".  Since we've received only the
            # "option_name(value)" part, we need to parse it.
            optname = option[:option.find("(")]
            optval = option[option.find("(") + 1:option.find(")")]

            val = ops["--" + optname]
            if val is None:
                val = ops["-" + optname]
            if val is None:
                return False
            return str(val) == optval or str(val) == optval
        else:
            return (ops["--" + option] is not None or
                    ops["-" + option] is not None)

    def match_date(self, date_arg, operator="date"):
        # The parsed scan's get_date() returns a time.struct_time, so we
        # need to convert it to a date object
        from datetime import date, datetime
        scd = self.parsed_scan.get_date()
        scan_date = date(scd.tm_year, scd.tm_mon, scd.tm_mday)

        # Check if we have any fuzzy operators ("~") in our string
        fuzz = 0
        if "~" in date_arg:
            # Count 'em, and strip 'em
            fuzz = date_arg.count("~")
            date_arg = date_arg.replace("~", "")

        if re.match(r"\d\d\d\d-\d\d-\d\d$", date_arg) is not None:
            year, month, day = date_arg.split("-")
            parsed_date = date(int(year), int(month), int(day))
        elif re.match(r"[-|\+]\d+$", date_arg):
            # We need to convert from the "-n" format (n days ago) to a date
            # object (I found this in some old code, don't ask :) )
            parsed_date = date.fromordinal(
                    date.today().toordinal() + int(date_arg))
        else:
            # Fail silently
            return False

        # Now that we have both the scan date and the user date converted to
        # date objects, we need to make a comparison based on the operator
        # (date, after, before).
        if operator == "date":
            return abs((scan_date - parsed_date).days) <= fuzz
        # We ignore fuzziness for after: and before:
        elif operator == "after":
            return (scan_date - parsed_date).days >= 0
        elif operator == "before":
            return (parsed_date - scan_date).days >= 0

    def match_after(self, date_arg):
        return self.match_date(date_arg, operator="after")

    def match_before(self, date_arg):
        return self.match_date(date_arg, operator="before")

    def match_target(self, target):
        log.debug("Match target: %s" % target)

        for spec in self.parsed_scan.get_targets():
            if target in spec:
                return True
        else:
            # We search the (rDNS) hostnames list
            for host in self.parsed_scan.get_hosts():
                if HostSearch.match_target(host, target):
                    return True
        return False

    def match_os(self, os):
        # If you have lots of big scans in your DB (with a lot of hosts
        # scanned), you're probably better off using the keyword (freetext)
        # search. Keyword search just greps through the nmap output, while this
        # function iterates through all parsed OS-related values for every host
        # in every scan!
        hosts = self.parsed_scan.get_hosts()
        for host in hosts:
            if HostSearch.match_os(host, os):
                return True
        return False

    def match_scanned(self, ports):
        if ports == "":
            return True

        # Transform a comma-delimited string containing ports into a list
        ports = [not_empty for not_empty in ports.split(",") if not_empty]

        # Check if they're parsable, if not return False silently
        for port in ports:
            if re.match(r"^\d+$", port) is None:
                return False

        # Make a list of all scanned ports
        services = []
        for scaninfo in self.parsed_scan.get_scaninfo():
            services.append(scaninfo["services"].split(","))

        # These two loops iterate over search ports and over scanned ports. As
        # soon as the search finds a given port among the scanned ports, it
        # breaks from the services loop and continues with the next port in the
        # ports list. If a port isn't found in the services list, the function
        # immediately returns False.
        for port in ports:
            for service in services:
                if "-" in service and \
                   int(port) >= int(service.split("-")[0]) and \
                   int(port) <= int(service.split("-")[1]):
                    # Port range, and our port was inside
                    break
                elif port == service:
                    break
            else:
                return False
        else:
            # The ports loop finished for all ports, which means the search was
            # successful.
            return True

    def match_port(self, ports, port_state):
        log.debug("Match port:%s" % ports)

        # Transform a comma-delimited string containing ports into a list
        ports = [not_empty for not_empty in ports.split(",") if not_empty]

        for host in self.parsed_scan.get_hosts():
            for port in ports:
                if not HostSearch.match_port(
                        host.get_ports(), port, port_state):
                    break
            else:
                return True
        else:
            return False

    def match_open(self, port):
        return self.match_port(port, "open")

    def match_filtered(self, port):
        return self.match_port(port, "filtered")

    def match_closed(self, port):
        return self.match_port(port, "closed")

    def match_unfiltered(self, port):
        return self.match_port(port, "unfiltered")

    def match_open_filtered(self, port):
        return self.match_port(port, "open|filtered")

    def match_closed_filtered(self, port):
        return self.match_port(port, "closed|filtered")

    def match_service(self, sversion):
        if sversion == "" or sversion == "*":
            return True

        for host in self.parsed_scan.get_hosts():
            if HostSearch.match_service(host, sversion):
                return True
        else:
            return False

    def match_in_route(self, host):
        if host == "" or host == "*":
            return True
        host = host.lower()

        # Since the parser doesn't parse traceroute output, we need to cheat
        # and look the host up in the Nmap output, in the Traceroute section of
        # the scan.
        nmap_out = self.parsed_scan.get_nmap_output()
        tr_pos = 0
        traceroutes = []        # A scan holds one traceroute section per host
        while tr_pos != -1:
            # Find the beginning and the end of the traceroute section, and
            # append the substring to the traceroutes list
            tr_pos = nmap_out.find("TRACEROUTE", tr_pos + 1)
            tr_end_pos = nmap_out.find("\n\n", tr_pos)
            if tr_pos != -1:
                traceroutes.append(nmap_out[tr_pos:tr_end_pos])

        for tr in traceroutes:
            if host in tr.lower():
                return True
        else:
            return False


class SearchDummy(SearchResult):
    """A dummy search class that returns no results. It is used as a
    placeholder when SearchDB can't be used."""
    def get_scan_results(self):
        return []


class SearchDB(SearchResult, object):
    def __init__(self):
        SearchResult.__init__(self)
        log.debug(">>> Getting scan results stored in data base")
        self.scan_results = []
        from zenmapCore.UmitDB import UmitDB
        u = UmitDB()

        for scan in u.get_scans():
            log.debug(">>> Retrieving result of scans_id %s" % scan.scans_id)
            log.debug(">>> Nmap xml output: %s" % scan.nmap_xml_output)

            try:
                buffer = io.StringIO(scan.nmap_xml_output)
                parsed = NmapParser()
                parsed.parse(buffer)
                buffer.close()
            except Exception as e:
                log.warning(">>> Error loading scan with ID %u from database: "
                        "%s" % (scan.scans_id, str(e)))
            else:
                self.scan_results.append(parsed)

    def get_scan_results(self):
        return self.scan_results


class SearchDir(SearchResult, object):
    def __init__(self, search_directory, file_extensions=["usr"]):
        SearchResult.__init__(self)
        log.debug(">>> SearchDir initialized")
        self.search_directory = search_directory

        if isinstance(file_extensions, str):
            self.file_extensions = file_extensions.split(";")
        elif isinstance(file_extensions, list):
            self.file_extensions = file_extensions
        else:
            raise Exception(
                    "Wrong file extension format! '%s'" % file_extensions)

        log.debug(">>> Getting directory's scan results")
        self.scan_results = []
        files = []
        for ext in self.file_extensions:
            files += glob(os.path.join(self.search_directory, "*.%s" % ext))

        log.debug(">>> Scan results at selected directory: %s" % files)
        for scan_file in files:
            log.debug(">>> Retrieving scan result %s" % scan_file)
            if os.access(scan_file, os.R_OK) and os.path.isfile(scan_file):

                try:
                    parsed = NmapParser()
                    parsed.parse_file(scan_file)
                except Exception:
                    pass
                else:
                    self.scan_results.append(parsed)

    def get_scan_results(self):
        return self.scan_results


class SearchResultTest(unittest.TestCase):
    class SearchClass(SearchResult):
        """This class is for use by the unit testing code"""
        def __init__(self, filenames):
            SearchResult.__init__(self)
            self.scan_results = []
            for filename in filenames:
                scan = NmapParser()
                scan.parse_file(filename)
                self.scan_results.append(scan)

        def get_scan_results(self):
            return self.scan_results

    def setUp(self):
        files = ["test/xml_test%d.xml" % no for no in range(1, 13)]
        self.search_result = self.SearchClass(files)

    def _test_skeleton(self, key, val):
        results = []
        search = {key: [val]}
        for scan in self.search_result.search(**search):
            results.append(scan)
        return len(results)

    def test_match_os(self):
        """Test that checks if the match_os predicate works"""
        assert(self._test_skeleton('os', 'linux') == 2)

    def test_match_target(self):
        """Test that checks if the match_target predicate works"""
        assert(self._test_skeleton('target', 'localhost') == 4)

    def test_match_port_open(self):
        """Test that checks if the match_open predicate works"""
        assert(self._test_skeleton('open', '22') == 7)

    def test_match_port_closed(self):
        """Test that checks if the match_closed predicate works"""
        assert(self._test_skeleton('open', '22') == 7)
        assert(self._test_skeleton('closed', '22') == 9)

    def test_match_service(self):
        """Test that checks if the match_service predicate works"""
        assert(self._test_skeleton('service', 'apache') == 9)
        assert(self._test_skeleton('service', 'openssh') == 7)

    def test_match_service_version(self):
        """Test that checks if the match_service predicate works when """
        """checking version"""
        assert(self._test_skeleton('service', '2.0.52') == 7)


if __name__ == "__main__":
    unittest.main()
