Сканируем случайные адреса Интернет. Получаем баннеры и протоколы с помощью Python

def ip_from_range(path_dir):
    """
    Получение случайного ip-адреса из файла с CIDR.

    Запускаем бесконечный цикл. Открываем файл с диапазонами
    cidr провайдеров. Выбираем случайный cidr, выбираем случайный ip.
    Проверяем, не использовался ли этот адрес ранее.
    Если нет, запускаем сканирование адреса. После, если словарь с результатом
    не пуст, сохраняем полученные данные. Если пуст, выводим сообщение пользователю,
    что нет результата. Вне зависимости от наполненности словаря сохраняем текущий ip,
    после чего очищаем словарь.
    :param path_dir: строка, путь к файлу с диапазонами (cidr).
    """
    global result
    while True:
        with open(path_dir, 'r', encoding='utf-8') as file:
            cidr = random.choice([x.strip() for x in file.readlines()])
            ip = random.choice([str(x) for x in ip_network(cidr)])
            if check_ip(ip):
                continue
            print(f"\nСканирую: {ip}")
            syn_ack_scan(ip, (1, 1001))
            if len(result) > 0:
                save_json(ip)
            else:
                print(f"No result\n{'-' * 65}")
            save_ip(ip)
            result.clear()

def scan_port(ip: str, port: int) -> (str, dict, bool):
    """
    Получение банера на отрытом потру.

    Выполняется подключение к ip-адресу на полученный
    порт. Если подключение удалось, значит порт открыт. Производиться
    попытка получить баннер. Если баннер не получен, обрабатывается
    исключение, в котором идет запрос службы на порту у функции за
    пределами класса. Если получить службу не удалось, возвращается
    unassigned. Полученные значения добавляются в словари для
    последующей обработки. В словарь с баннерами добавляются значения,
    если баннер получен. В остальных случаях, в словарь открытых портов.

    :param ip: ip-адрес или домен для сканирования.
    :param port: Номер порта для сканирования. Целое число
    :return: Возвращается False. Служит для выхода из функции
    """
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(5)

    try:
        s.connect((ip, port))
        try:
            banner = s.recv(1024).decode().strip()
            if banner == '':
                return False
            else:
                return banner
        except OSError:
            return False
        except UnicodeDecodeError:
            banner = s.recv(1024).strip()
            return banner
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False

def syn_ack_scan(ip: str, ports: tuple):
    # создание пакета для сканирования
    try:
        request_syn = sc.IP(dst=ip) / sc.TCP(dport=ports, flags="S")
    except socket.gaierror:
        raise ValueError(f'{ip} получить не удалось')
    answer = sc.sr(request_syn, timeout=2, retry=1, verbose=False)[0]  # отправка пакета

    for _, receive in answer:
        try:
            if receive['TCP'].flags == "SA":
                try:
                    if ip not in result:
                        result[ip] = dict()
                    if receive['TCP'].sport == 80:
                        try:
                            head = requests.head(url=f"https://{ip}:80", headers=headers, verify=False,
                                                 timeout=3).headers
                            try:
                                result[ip].update({'80/http': case_sence(head)['location']})
                            except Exception:
                                result[ip].update({'80/http': case_sence(head)})
                        except Exception:
                            result[ip].update({"80/http": "open"})
                    elif receive['TCP'].sport == 443:
                        try:
                            head = requests.head(url=f"https://{ip}:443", headers=headers, verify=False,
                                                 timeout=3).headers
                            try:
                                result[ip].update({"443/http": case_sence(head)['location']})
                            except Exception:
                                result[ip].update({"443/http": case_sence(head)})
                        except Exception:
                            result[ip].update({"443/http": "open"})
                    else:
                        resp = scan_port(ip=ip, port=receive['TCP'].sport)
                        if not resp:
                            result[ip].update({f"{str(receive['TCP'].sport)}/"
                                               f"{sc.TCP_SERVICES[receive['TCP'].sport]}": "open"})
                        else:
                            result[ip].update({f"{str(receive['TCP'].sport)}/"
                                               f"{sc.TCP_SERVICES[receive['TCP'].sport]}": resp})
                except KeyError:
                    result[ip].update({f"{str(receive['TCP'].sport)}/undefined": "open"})
        except IndexError:
            continue

"""
Скрипт для сканирования ip-адресов выбранных случайно из
списка CIDR провайдеров.

Для работы скрипта требуется установка библиотек:
pip install --pre scapy[basic] requests
"""

import json
import random
import socket
import sys
from datetime import datetime as dt
from ipaddress import ip_network
from os import getuid
from pathlib import Path
from platform import system

import requests
import scapy.all as sc

headers = {
    "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 "
                  "Safari/537.36 "
}

requests.packages.urllib3.disable_warnings()
random.seed(dt.now().second + random.randint(0, 101))

result = dict()


def scan_port(ip: str, port: int) -> (str, dict, bool):
    """
    Получение банера на отрытом потру.

    Выполняется подключение к ip-адресу на полученный
    порт. Если подключение удалось, значит порт открыт. Производиться
    попытка получить баннер. Если баннер не получен, обрабатывается
    исключение, в котором идет запрос службы на порту у функции за
    пределами класса. Если получить службу не удалось, возвращается
    unassigned. Полученные значения добавляются в словари для
    последующей обработки. В словарь с баннерами добавляются значения,
    если баннер получен. В остальных случаях, в словарь открытых портов.

    :param ip: ip-адрес или домен для сканирования.
    :param port: Номер порта для сканирования. Целое число
    :return: Возвращается False. Служит для выхода из функции
    """
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(5)

    try:
        s.connect((ip, port))
        try:
            banner = s.recv(1024).decode().strip()
            if banner == '':
                return False
            else:
                return banner
        except OSError:
            return False
        except UnicodeDecodeError:
            banner = s.recv(1024).strip()
            return banner
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False


def syn_ack_scan(ip: str, ports: tuple):
    """
    Сканирование портов в переданном диапазоне на переданном ip-адресе.

    Выполняется сканирование портов в указанном диапазоне. Если есть ответ
    по текущему порту, выполняется попытка получения банера, если это не 80 и 443 порты.
    Если это 80 или 443 порт, пытаемся получить заголовки сайта, который возможно
    хостится на данном сервере по-данному ip.
    Дабавляем все полученные данные в глобальный словарь.

    :param ip: строка, ip-адрес цели.
    :param ports: кортеж из портов для сканирования.
    """
    # создание пакета для сканирования
    try:
        request_syn = sc.IP(dst=ip) / sc.TCP(dport=ports, flags="S")
    except socket.gaierror:
        raise ValueError(f'{ip} получить не удалось')
    answer = sc.sr(request_syn, timeout=2, retry=1, verbose=False)[0]  # отправка пакета

    for _, receive in answer:
        try:
            if receive['TCP'].flags == "SA":
                try:
                    print(receive['TCP'].sport)
                    if ip not in result:
                        result[ip] = dict()
                    if receive['TCP'].sport == 80:
                        try:
                            head = requests.head(url=f"https://{ip}:80", headers=headers, verify=False,
                                                 timeout=3).headers
                            try:
                                result[ip].update({'80/http': case_sence(head)['location']})
                            except Exception:
                                result[ip].update({'80/http': case_sence(head)})
                        except Exception:
                            result[ip].update({"80/http": "open"})
                    elif receive['TCP'].sport == 443:
                        try:
                            head = requests.head(url=f"https://{ip}:443", headers=headers, verify=False,
                                                 timeout=3).headers
                            try:
                                result[ip].update({"443/http": case_sence(head)['location']})
                            except Exception:
                                result[ip].update({"443/http": case_sence(head)})
                        except Exception:
                            result[ip].update({"443/http": "open"})
                    else:
                        resp = scan_port(ip=ip, port=receive['TCP'].sport)
                        if not resp:
                            result[ip].update({f"{str(receive['TCP'].sport)}/"
                                               f"{sc.TCP_SERVICES[receive['TCP'].sport]}": "open"})
                        else:
                            result[ip].update({f"{str(receive['TCP'].sport)}/"
                                               f"{sc.TCP_SERVICES[receive['TCP'].sport]}": resp})
                except KeyError:
                    result[ip].update({f"{str(receive['TCP'].sport)}/undefined": "open"})
        except IndexError:
            continue


def case_sence(resp: dict) -> dict:
    """
    Перевод регистра ключей и значений в нижний диапазон,
    так как запись в верхнем регистре в json невозможна.

    :param resp: словарь с данными для перевода.
    :return: словарь с переведенными данными.
    """
    resp_low = dict()
    for res in resp:
        resp_low.update({res.lower(): resp[res].lower()})
    return resp_low


def save_json(addr):
    """
    Сохранение данных из словаря с полученными значениями.

    Создаем директорию для сохранения json, если она не существует.
    Забираем дату в нужном формате. Проверяем, не является ли словарь result пустым.
    Записываем значения из словаря в файл json.
    Выводим информацию из словаря в терминал для пользователя.

    :param addr: ip-адрес, по которому выполнялось текущее сканирование.
    """
    (Path.cwd() / "checked_ip").mkdir(exist_ok=True)

    day = dt.now().strftime("%d-%m-%Y")
    if len(result) > 0:
        with open(Path.cwd() / "checked_ip" / f"{day}_{addr}.json", 'w', encoding='utf-8') as file:
            json.dump(result, file, indent=4, ensure_ascii=False)
        for res in result:
            print(f"\nИнформация об адресе: {res}\n{'-'*65}")
            for st in result[res]:
                print(f"Порт: {st:20} | Статус: {result[res][st]}")
        print(f"{'-'*65}")


def save_ip(ip):
    """
    Сохраняет текущие ip-адреса в текстовый документ, для
    последующей проверки по ним, перед сканированием, чтобы
    избежать повторного сканирования.

    Открываем файл, дописываем адрес в конец файла.
    :param ip: ip-адрес для записи в файл.
    """
    with open(Path.cwd() / "used_ip.txt", 'a', encoding='utf-8') as file:
        file.write(f"{ip}\n")


def check_ip(ip):
    """
    Проверка ip-адреса на нахождение в списке уже сканированных адресов.

    Проверяем, существует ли файл. Открываем, создаем список из адресов
    и проверяем вхождение текущего адреса в сформированный список.
    Если есть, возвращаем True, нет - False.
    :param ip: строка, ip-адрес для проверки.
    :return: булево значение. True или False в зависимости от результата.
    """
    if (Path.cwd() / "used_ip.txt").exists():
        with open(Path.cwd() / "used_ip.txt", 'r', encoding='utf-8') as file:
            lines = [x.strip() for x in file.readlines()]
            if ip in lines:
                return True
    return False


def ip_from_range(path_dir):
    """
    Получение случайного ip-адреса из файла с CIDR.

    Запускаем бесконечный цикл. Открываем файл с диапазонами
    cidr провайдеров. Выбираем случайный cidr, выбираем случайный ip.
    Проверяем, не использовался ли этот адрес ранее.
    Если нет, запускаем сканирование адреса. После, если словарь с результатом
    не пуст, сохраняем полученные данные. Если пуст, выводим сообщение пользователю,
    что нет результата. Вне зависимости от наполненности словаря сохраняем текущий ip,
    после чего очищаем словарь.
    :param path_dir: строка, путь к файлу с диапазонами (cidr).
    """
    global result
    while True:
        with open(path_dir, 'r', encoding='utf-8') as file:
            cidr = random.choice([x.strip() for x in file.readlines()])
            ip = random.choice([str(x) for x in ip_network(cidr)])
            if check_ip(ip):
                continue
            print(f"\nСканирую: {ip}")
            syn_ack_scan(ip, (1, 1001))
            if len(result) > 0:
                save_json(ip)
            else:
                print(f"No result\n{'-' * 65}")
            save_ip(ip)
            result.clear()


def main():
    """
    Запуск сканирования адресов.
    """
    if system() == "Linux":
        if not getuid() == 0:
            print("\n [+] Run the script as root user!")
            sys.exit(0)

    ip_from_range("provider_ranges.txt")


if __name__ == "__main__":
    main()