Проверка безопасности Elasticsearch, или «я вижу ваши индексы»

#!/usr/bin/env python3

import ipaddress

import socket

import json

from urllib.request import Request, urlopen

from urllib.error import URLError, HTTPError

NETWORK = "192.168.56.0/24"  # замените на свой учебный или корпоративный диапазон

PORT = 9200

TIMEOUT = 1.5

def is_port_open(host: str, port: int) -> bool:

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:

        sock.settimeout(TIMEOUT)

        return sock.connect_ex((host, port)) == 0

def check_elasticsearch(host: str):

    url = f"http://{host}:{PORT}/"

    request = Request(url, headers={"User-Agent": "internal-es-audit/1.0"})

    try:

        with urlopen(request, timeout=TIMEOUT) as response:

            body = response.read(4096).decode("utf-8", errors="replace")

            data = json.loads(body)

            if "cluster_name" in data or "tagline" in data:

                return {

                    "host": host,

                    "url": url,

                    "cluster_name": data.get("cluster_name"),

                    "version": data.get("version", {}).get("number"),

                    "tagline": data.get("tagline"),

                }

    except (URLError, HTTPError, TimeoutError, json.JSONDecodeError):

        return None

    return None

def main():

    network = ipaddress.ip_network(NETWORK, strict=False)

    for ip in network.hosts():

        host = str(ip)

        if not is_port_open(host, PORT):

            continue

        result = check_elasticsearch(host)

        if result:

            print(json.dumps(result, ensure_ascii=False, indent=2))

        else:

            print(f"[?] {host}:{PORT} открыт, но ответ не похож на Elasticsearch")

if __name__ == "__main__":

    main()

#!/usr/bin/env python3

import argparse

import json

from getpass import getpass

from urllib.request import Request, urlopen

from urllib.error import HTTPError, URLError

import base64

def build_auth_header(username: str | None, password: str | None):

    if not username:

        return {}

    token = f"{username}:{password}".encode("utf-8")

    encoded = base64.b64encode(token).decode("ascii")

    return {"Authorization": f"Basic {encoded}"}

def fetch_indices(base_url: str, username: str | None, password: str | None):

    url = base_url.rstrip("/") + "/_cat/indices?format=json&bytes=mb"

    headers = {

        "User-Agent": "internal-es-index-audit/1.0",

        **build_auth_header(username, password),

    }

    request = Request(url, headers=headers)

    with urlopen(request, timeout=5) as response:

        return json.loads(response.read().decode("utf-8"))

def main():

    parser = argparse.ArgumentParser(

        description="List Elasticsearch indices for authorized security checks."

    )

    parser.add_argument("url", help="Example: http://192.168.56.12:9200")

    parser.add_argument("-u", "--username", help="Elasticsearch username")

    args = parser.parse_args()

    password = getpass("Password: ") if args.username else None

    try:

        indices = fetch_indices(args.url, args.username, password)

        print(f"{'health':<8} {'status':<8} {'index':<40} {'docs.count':>12} {'store.size':>12}")

        print("-" * 88)

        for item in indices:

            print(

                f"{item.get('health', ''):<8} "

                f"{item.get('status', ''):<8} "

                f"{item.get('index', ''):<40} "

                f"{item.get('docs.count', ''):>12} "

                f"{item.get('store.size', ''):>12}"

            )

    except HTTPError as error:

        print(f"HTTP error: {error.code} {error.reason}")

    except URLError as error:

        print(f"Connection error: {error.reason}")

    except json.JSONDecodeError:

        print("Response was not valid JSON")

if __name__ == "__main__":

    main()