<commit_before>import abc
import os
import threading
from collections import namedtuple
from typing import List, Tuple, NamedTuple, Set

from analyzr import constants, InvalidInterface
from analyzr.constants import NUM_PING_THREADS, TCPFlag, IPFlag, topports
from analyzr.fingerprinters import Fingerprinter

DiscoveredHost = namedtuple("DiscoveredHost", ["ip", "mac"])
PingedHost = namedtuple("PingedHost", ["ip", "ttl"])
HostInfo = NamedTuple("HostInfo", [("ip", str), ("ip_network", str), ("gateway_ip", str), ("dhcp_server_ip", str),
                                   ("interface_name", str)])


class NetworkToolFacade(object):
    __metaclass__ = abc.ABCMeta

    def __init__(self, fingerprinters: List[Fingerprinter], interface_to_use: str = None):
        """
        Inits the object. Raises InvalidInterface exception if given interface is invalid (maps to no interface
        on the host).

        :param fingerprinters: List of fingerprinters to use to identify packet.
        :param interface_to_use: the network interface used by the tool
        """
        self.fingerprinters = fingerprinters
        self.interface_to_use = interface_to_use

    @property
    @abc.abstractmethod
    def host_information(self) -> HostInfo:
        pass

    @abc.abstractmethod
    def arp_discover_hosts(self,
                           network: str,
                           timeout: int,
                           verbose: bool = False) -> List[DiscoveredHost]:
        """
        Returns list of DiscoveredHost
        :param network:
        :param timeout:
        :param verbose:
        """
        pass

    @abc.abstractmethod
    def passive_discover_hosts(self,
                               networks: List[str],
                               timeout_in_secs: int) -> List[DiscoveredHost]:
        """

        :param networks: Only check for traffic that comes from these networks.
                         Networks must be in CIRD format (I.E: 192.168.1.0/24)

        :param timeout_in_secs: Number of seconds to passively scan the network before stopping.
        """

    @abc.abstractmethod
    def route_to_target(self, target_ip: str):
        """
        Returns route to target, or an empty list if no route found.

        :param target_ip: target ip address
        :return: A list containing the trace of IPs to get to target.
        """
        pass

    @abc.abstractmethod
    def icmp_ping(self, ip: str, timeout: int, verbose: bool) -> PingedHost:
        """
        If host exists, will return a PingedHost namedtuple. Else returns None.

        :param ip: The ip to icmp_ping
        :param timeout:  Time before giving up
        :param verbose: If set to True, there may be logs in stdout.
        """
        pass

    @abc.abstractmethod
    def identify_host_os(self, ip: str) -> Set[str]:
        """
        Tries to identify the operating system of the host at the given IP.

        Returns None if nothing found. Else return a set of possible matches.

        :param ip: The host to probe
        """
        pass

    @abc.abstractmethod
    def tcp_port_scan(self, ip: str, ports_to_scan: List[int]) -> Tuple[List[int], List[int]]:
        """
        Checks each port of ports_to_scan to see if it is opened on the host.

        Returns tuple: list of opened ports (sorted in order), list of closed ports (sorted in order).
        Never returns None. Both lists will always be at least empty.

        :param ip: The ip to scan for opened ports
        :param ports_to_scan: The ports to scan on the host.

        :returns tuple opened_ports, closed_ports.
        """
        pass

    @staticmethod
    @abc.abstractmethod
    def pkt_to_ettercap_fn() -> callable:
        """
        Returns a function that converts a pkt for processing by a fingerprinter.
        """
        pass

    @abc.abstractmethod
    def sniff_network(self,
                      nb_pkts_to_sniff: int,
                      store_sniffed_pkts: bool,
                      timeout_in_secs: int = None,
                      pkt_callback_fn: callable = None,
                      pkt_filter_fn: callable = None,
                      stop_filter_fn: callable = None) -> list:
        pass


with open(os.devnull, 'w') as f:
    try:
        from contextlib import redirect_stderr
    except ImportError:
        from analyzr.utils.context import redirect_stderr
    with redirect_stderr(f):
        # stderr redirected to os.devnull. No annoying import messages
        # printed on module import
        from scapy.all import *

from scapy.layers.dhcp import DHCP, BOOTP
from scapy.layers.inet import IP, ICMP, TCP, traceroute
from scapy.layers.inet import UDP


class ScapyTool(NetworkToolFacade):
    # Removes "WARNING: Mac address to reach destination not found. Using broadcast" message.
    logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
    cached_host_info = None

    scapy_tcp_flags = {"SYN": "S", "ACK": "A", "RST": "R"}

    def __init__(self, fingerprinters: List[Fingerprinter], interface_to_use: str = None):
        from scapy.all import conf
        # Make scapy shut up while sending packets
        conf.verb = 0

        if not interface_to_use:
            interface_to_use = conf.iface

        super().__init__(fingerprinters, interface_to_use)

        self.logger = logging.getLogger(__name__)<commit_msg>Lorsque l'interface est "auto", ScapyTool prends l'interface défini par Scapy.<commit_after>import abc
import os
import threading
from collections import namedtuple
from typing import List, Tuple, NamedTuple, Set

from analyzr import constants, InvalidInterface
from analyzr.constants import NUM_PING_THREADS, TCPFlag, IPFlag, topports
from analyzr.fingerprinters import Fingerprinter

DiscoveredHost = namedtuple("DiscoveredHost", ["ip", "mac"])
PingedHost = namedtuple("PingedHost", ["ip", "ttl"])
HostInfo = NamedTuple("HostInfo", [("ip", str), ("ip_network", str), ("gateway_ip", str), ("dhcp_server_ip", str),
                                   ("interface_name", str)])


class NetworkToolFacade(object):
    __metaclass__ = abc.ABCMeta

    def __init__(self, fingerprinters: List[Fingerprinter], interface_to_use: str = None):
        """
        Inits the object. Raises InvalidInterface exception if given interface is invalid (maps to no interface
        on the host).

        :param fingerprinters: List of fingerprinters to use to identify packet.
        :param interface_to_use: the network interface used by the tool. if "auto" or None, then it is automagically
        selected (which one is chosen depends on the NetworkTool, but best bet is on the primary network interface).
        """
        self.fingerprinters = fingerprinters
        self.interface_to_use = interface_to_use

    @property
    @abc.abstractmethod
    def host_information(self) -> HostInfo:
        pass

    @abc.abstractmethod
    def arp_discover_hosts(self,
                           network: str,
                           timeout: int,
                           verbose: bool = False) -> List[DiscoveredHost]:
        """
        Returns list of DiscoveredHost
        :param network:
        :param timeout:
        :param verbose:
        """
        pass

    @abc.abstractmethod
    def passive_discover_hosts(self,
                               networks: List[str],
                               timeout_in_secs: int) -> List[DiscoveredHost]:
        """

        :param networks: Only check for traffic that comes from these networks.
                         Networks must be in CIRD format (I.E: 192.168.1.0/24)

        :param timeout_in_secs: Number of seconds to passively scan the network before stopping.
        """

    @abc.abstractmethod
    def route_to_target(self, target_ip: str):
        """
        Returns route to target, or an empty list if no route found.

        :param target_ip: target ip address
        :return: A list containing the trace of IPs to get to target.
        """
        pass

    @abc.abstractmethod
    def icmp_ping(self, ip: str, timeout: int, verbose: bool) -> PingedHost:
        """
        If host exists, will return a PingedHost namedtuple. Else returns None.

        :param ip: The ip to icmp_ping
        :param timeout:  Time before giving up
        :param verbose: If set to True, there may be logs in stdout.
        """
        pass

    @abc.abstractmethod
    def identify_host_os(self, ip: str) -> Set[str]:
        """
        Tries to identify the operating system of the host at the given IP.

        Returns None if nothing found. Else return a set of possible matches.

        :param ip: The host to probe
        """
        pass

    @abc.abstractmethod
    def tcp_port_scan(self, ip: str, ports_to_scan: List[int]) -> Tuple[List[int], List[int]]:
        """
        Checks each port of ports_to_scan to see if it is opened on the host.

        Returns tuple: list of opened ports (sorted in order), list of closed ports (sorted in order).
        Never returns None. Both lists will always be at least empty.

        :param ip: The ip to scan for opened ports
        :param ports_to_scan: The ports to scan on the host.

        :returns tuple opened_ports, closed_ports.
        """
        pass

    @staticmethod
    @abc.abstractmethod
    def pkt_to_ettercap_fn() -> callable:
        """
        Returns a function that converts a pkt for processing by a fingerprinter.
        """
        pass

    @abc.abstractmethod
    def sniff_network(self,
                      nb_pkts_to_sniff: int,
                      store_sniffed_pkts: bool,
                      timeout_in_secs: int = None,
                      pkt_callback_fn: callable = None,
                      pkt_filter_fn: callable = None,
                      stop_filter_fn: callable = None) -> list:
        pass


with open(os.devnull, 'w') as f:
    try:
        from contextlib import redirect_stderr
    except ImportError:
        from analyzr.utils.context import redirect_stderr
    with redirect_stderr(f):
        # stderr redirected to os.devnull. No annoying import messages
        # printed on module import
        from scapy.all import *

from scapy.layers.dhcp import DHCP, BOOTP
from scapy.layers.inet import IP, ICMP, TCP, traceroute
from scapy.layers.inet import UDP


class ScapyTool(NetworkToolFacade):
    # Removes "WARNING: Mac address to reach destination not found. Using broadcast" message.
    logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
    cached_host_info = None

    scapy_tcp_flags = {"SYN": "S", "ACK": "A", "RST": "R"}

    def __init__(self, fingerprinters: List[Fingerprinter], interface_to_use: str = None):
        from scapy.all import conf
        # Make scapy shut up while sending packets
        conf.verb = 0

        if not interface_to_use or interface_to_use == "auto":
            interface_to_use = conf.iface

        super().__init__(fingerprinters, interface_to_use)

        self.logger = logging.getLogger(__name__)
{'fraction_non_alphanumeric': 0.0799275033982782, 'fraction_numerical': 0.001993656547349343, 'mean_word_length': 2.321095395726753, 'pattern_counts': {'":': 7, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1964427', 'n_tokens_mistral': 3129, 'n_tokens_neox': 3031, 'n_words': 1047}