OT/ICS Shodan Searches

port:502 country:AU "Schneider Electric"

port:102 "Siemens" 
port:44818 "Allen-Bradley"
port:502 "Modicon"
port:1911 "Niagara"
port:4840 "OPC UA"


port:47808
port:20000
port:2404
port:4840
port:1911

#!/usr/bin/env python3

import argparse
import socket
import sys
from dataclasses import dataclass


@dataclass
class Vulnerability:
    cve: str
    description: str
    affected_before: str | None = None
    affected_all_versions: bool = False


VULNERABILITIES = [
    Vulnerability(
        "CVE-2017-6030",
        "Predictable TCP initial sequence numbers",
        affected_before="1.5.0.0",
    ),
    Vulnerability(
        "CVE-2017-7574",
        "Hard-coded cryptographic key in project protection",
        affected_before="1.5.0.0",
    ),
    Vulnerability(
        "CVE-2017-7575",
        "Weak application-protection authentication",
        affected_before="1.5.0.2",
    ),
    Vulnerability(
        "CVE-2018-7789",
        "Improper handling of unusual/exceptional conditions",
        affected_before="1.6.2.0",
    ),
    Vulnerability(
        "CVE-2018-7790",
        "Authentication sequence replay",
        affected_before="1.6.2.0",
    ),
    Vulnerability(
        "CVE-2018-7791",
        "Improper access control",
        affected_before="1.6.2.0",
    ),
    Vulnerability(
        "CVE-2018-7792",
        "Missing authorization / weak cryptography",
        affected_before="1.6.2.0",
    ),
    Vulnerability(
        "CVE-2018-7798",
        "Insufficient verification of network configuration data",
        affected_all_versions=True,
    ),
    Vulnerability(
        "CVE-2018-7821",
        "Resource exhaustion involving Ethernet/IP",
        affected_before="1.10.0.0",
    ),
    Vulnerability(
        "CVE-2018-7822",
        "Incorrect default permissions",
        affected_before="1.10.0.0",
    ),
    Vulnerability(
        "CVE-2018-7823",
        "M221/SoMachine Basic security vulnerability",
        affected_before="1.10.0.0",
    ),
    Vulnerability(
        "CVE-2019-6820",
        "Unauthenticated modification of IP configuration",
        affected_all_versions=True,
    ),
    Vulnerability(
        "CVE-2020-28214",
        "Predictable salt in password hashing",
        affected_all_versions=True,
    ),
]


def parse_version(version):
    try:
        return tuple(int(part) for part in version.split("."))
    except ValueError as exc:
        raise ValueError(
            f"Invalid firmware version: {version}"
        ) from exc


def version_is_less_than(current, target):
    return parse_version(current) < parse_version(target)


def check_tcp_port(host, port, timeout):
    try:
        with socket.create_connection(
            (host, port),
            timeout=timeout,
        ):
            return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False


def assess_vulnerabilities(firmware):
    results = []

    for vulnerability in VULNERABILITIES:
        if vulnerability.affected_all_versions:
            results.append(
                (
                    vulnerability.cve,
                    "POTENTIALLY AFFECTED",
                    vulnerability.description,
                )
            )
            continue

        if version_is_less_than(
            firmware,
            vulnerability.affected_before,
        ):
            results.append(
                (
                    vulnerability.cve,
                    "POTENTIALLY AFFECTED",
                    vulnerability.description,
                )
            )
        else:
            results.append(
                (
                    vulnerability.cve,
                    "NOT AFFECTED BY VERSION RANGE",
                    vulnerability.description,
                )
            )

    return results


def print_results(host, firmware, port_open, results):
    print()
    print("Modicon M221 Safe Vulnerability Assessment")
    print("=" * 48)
    print(f"Target:   {host}")
    print(f"Firmware: {firmware}")
    print(f"TCP/502:  {'OPEN' if port_open else 'CLOSED/UNREACHABLE'}")
    print()

    print("CVE Assessment")
    print("-" * 48)

    for cve, status, description in results:
        print(f"{cve:<18} {status}")
        print(f"  {description}")

    print()
    print("Safety limitations")
    print("-" * 48)
    print("No authentication bypass was attempted.")
    print("No passwords were requested or recovered.")
    print("No PLC program was downloaded.")
    print("No PLC configuration was modified.")
    print("No registers or coils were written.")
    print("No denial-of-service or packet-flooding test was performed.")
    print()
    print(
        "CVE results are firmware/version assessments and should be "
        "treated as potentially affected until verified against the "
        "vendor advisory and device configuration."
    )


def main():
    parser = argparse.ArgumentParser(
        description=(
            "Safe Modicon M221 vulnerability assessment. "
            "Performs TCP/502 connectivity testing and "
            "firmware-based CVE assessment."
        )
    )

    parser.add_argument(
        "target",
        help="M221 IP address or hostname",
    )

    parser.add_argument(
        "--firmware",
        required=True,
        help="Confirmed M221 firmware version, for example 1.3.3.3",
    )

    parser.add_argument(
        "--timeout",
        type=float,
        default=3.0,
        help="TCP connection timeout in seconds",
    )

    args = parser.parse_args()

    try:
        parse_version(args.firmware)
    except ValueError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(2)

    print(f"Checking {args.target}:502...")

    port_open = check_tcp_port(
        args.target,
        502,
        args.timeout,
    )

    results = assess_vulnerabilities(args.firmware)

    print_results(
        args.target,
        args.firmware,
        port_open,
        results,
    )


if __name__ == "__main__":
    main()

import socket
import sys


def check_modbus(ip, port=502, timeout=3):
    request = bytes.fromhex(
        "000100000006010300000001"
    )

    try:
        with socket.create_connection((ip, port), timeout=timeout) as sock:
            sock.sendall(request)
            response = sock.recv(256)

        if len(response) >= 9 and response[7] == 0x03:
            print(f"[+] {ip}:{port} responded to Modbus/TCP")
            print(f"[+] Response: {response.hex()}")
            return True

        print(f"[-] {ip}:{port} responded, but not as expected")
        print(f"[-] Response: {response.hex()}")

    except (socket.timeout, ConnectionRefusedError, OSError) as exc:
        print(f"[-] Unable to connect to {ip}:{port}: {exc}")

    return False


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <PLC-IP>")
        sys.exit(1)

    check_modbus(sys.argv[1])