initial

5 files changed, 294 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..0bdec20
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,2 @@
+/.idea
+/venv
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..6f8f585
--- /dev/null
+++ b/README.md
@@ -0,0 +1,51 @@
+# suddenly-opened-ports-checker
+
+Python script that scans TCP ports of your servers and notifies you about
+unexpected changes (new opened ports, or closed ports expected to be open).
+
+## Usage
+
+Python 3.7 or newer is required.
+
+```
+usage: suddenly-opened-ports-checker.py [-h] --config CONFIG [--verbose] [--concurrency CONCURRENCY] [--timeout TIMEOUT] [--threads-limit THREADS_LIMIT] [--no-telegram]
+
+optional arguments:
+  -h, --help            show this help message and exit
+  --config CONFIG       path to config file in yaml format
+  --verbose             set logging level to DEBUG (default is INFO)
+  --concurrency CONCURRENCY
+                        default number of threads per target (defaults to 200)
+  --timeout TIMEOUT     default timeout (defaults to 5)
+  --threads-limit THREADS_LIMIT
+                        global threads limit (default is no limit)
+  --no-telegram         just print results, don't send to telegram
+
+```
+
+## Config example
+
+Each server definition must have at least `host` and `opened` keys. `opened` is
+a list of ports expected to be open.
+
+You can also set per-server `concurrency` and `timeout`.
+
+```yaml
+server-1:
+  host: 1.2.3.4
+  opened:
+    - 22
+    - 80
+    - 443
+
+server-2:
+  host: 5.6.7.8
+  opened: []
+  concurrency: 1000
+  timeout: 2
+```
+
+## License
+
+MIT
+``
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..27c14b4
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,2 @@
+PyYAML~=5.4.1
+ch1p~=0.0.5
\ No newline at end of file
diff --git a/scanner.py b/scanner.py
new file mode 100644
index 0000000..73c82bd
--- /dev/null
+++ b/scanner.py
@@ -0,0 +1,76 @@
+import struct
+import socket
+import threading
+import queue
+import logging
+
+from enum import Enum, auto
+
+logger = logging.getLogger(__name__)
+
+
+class PortState(Enum):
+    OPEN = auto()
+    CLOSED = auto()
+    FILTERED = auto()
+
+
+class TCPScanner:
+    def __init__(self, host, ports, timeout=5):
+        self.host = host
+        self.ports = ports
+        self.timeout = timeout
+        self.results = []
+        self.q = queue.SimpleQueue()
+
+    def scan(self, num_threads=5):
+        for port in self.ports:
+            self.q.put(port)
+
+        threads = []
+        for i in range(num_threads):
+            t = threading.Thread(target=self.run)
+            t.start()
+            threads.append(t)
+
+        for t in threads:
+            t.join()
+
+        return self.results
+
+    def run(self):
+        try:
+            while True:
+                self._scan(self.q.get(block=False))
+        except queue.Empty:
+            return
+
+    def _scan(self, port):
+        try:
+            conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+            conn.settimeout(self.timeout)
+
+            ret = conn.connect_ex((self.host, port))
+
+            # DATA RECEIVED - SYN ACK
+            if ret == 0:
+                logger.debug('%s:%d - tcp open (SYN-ACK packet)' % (self.host, port))
+                self.results.append((port, PortState.OPEN))
+
+            # RST RECEIVED - PORT CLOSED
+            elif ret == 111:
+                logger.debug('%s:%d - tcp closed (RST packet)' % (self.host, port))
+                self.results.append((port, PortState.CLOSED))
+
+            # ERR CODE 11 - TIMEOUT
+            elif ret == 11:
+                self.results.append((port, PortState.FILTERED))
+
+            else:
+                logger.debug('%s:%d - code %d' % (self.host, port, ret))
+
+            conn.close()
+
+        except socket.timeout:
+            self.results.append((port, PortState.FILTERED))
diff --git a/suddenly-opened-ports-checker.py b/suddenly-opened-ports-checker.py
new file mode 100755
index 0000000..3c25ab6
--- /dev/null
+++ b/suddenly-opened-ports-checker.py
@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+import logging
+import yaml
+import math
+
+from argparse import ArgumentParser
+from ch1p import telegram_notify
+from threading import Thread, Lock
+from html import escape
+from scanner import TCPScanner, PortState
+
+mutex = Lock()
+logger = logging.getLogger(__name__)
+
+
+class Colored:
+    GREEN = '\033[92m'
+    RED = '\033[91m'
+    END = '\033[0m'
+
+
+class Results:
+    def __init__(self):
+        self.warnings = []
+        self.mutex = Lock()
+
+    def add(self, worker):
+        host = worker.get_host()
+        with self.mutex:
+            if not worker.done:
+                print(f'{Colored.RED}{worker.name}: scanning failed{Colored.END}')
+                return
+
+            print(f'{worker.name} ({host}):')
+
+            opened = []
+            results = worker.get_results()
+            for port, state in results:
+                if state != PortState.OPEN:
+                    continue
+
+                opened.append(port)
+                if not worker.is_expected(port):
+                    self.warnings.append(f'On {worker.name} ({host}): port {port} is open')
+                    print(f'    {Colored.RED}{port} opened{Colored.END}')
+                else:
+                    print(f'    {Colored.GREEN}{port} opened{Colored.END}')
+
+            if worker.opened:
+                for port in worker.opened:
+                    if port not in opened:
+                        self.warnings.append(
+                            f'On {worker.name} ({host}): port {port} expected to be opened, but it\'s not')
+                        print(f'    {Colored.RED}{port} not opened{Colored.END}')
+            print()
+
+    def has_warnings(self):
+        return len(self.warnings) > 0
+
+    def notify(self):
+        telegram_notify(escape('\n'.join(self.warnings)), parse_mode='html')
+
+
+class Worker(Thread):
+    def __init__(self, name, host, opened=None, concurrency=None, timeout=None):
+        Thread.__init__(self)
+
+        assert concurrency is not None
+
+        self.done = False
+        self.name = name
+        self.concurrency = concurrency
+        self.opened = opened
+
+        scanner_kw = {}
+        if timeout is not None:
+            scanner_kw['timeout'] = timeout
+        self.scanner = TCPScanner(host, range(0, 65535), **scanner_kw)
+
+    def run(self):
+        logger.info(f'starting {self.name} ({self.concurrency} threads)')
+        self.scanner.scan(num_threads=self.concurrency)
+        self.done = True
+        logger.info(f'finished {self.name}')
+
+    def get_results(self):
+        return self.scanner.results
+
+    def is_expected(self, port):
+        return (self.opened is not None) and (port in self.opened)
+
+    def get_host(self):
+        return self.scanner.host
+
+
+def main():
+    parser = ArgumentParser()
+    parser.add_argument('--config', type=str, required=True,
+                        help='path to config file in yaml format')
+    parser.add_argument('--verbose', action='store_true',
+                        help='set logging level to DEBUG')
+    parser.add_argument('--concurrency', default=200, type=int,
+                        help='default number of threads per target')
+    parser.add_argument('--timeout', default=5, type=int,
+                        help='default timeout')
+    parser.add_argument('--threads-limit', default=0, type=int,
+                        help='global threads limit')
+    parser.add_argument('--no-telegram', action='store_true',
+                        help='just print results, don\'t send to telegram')
+    args = parser.parse_args()
+
+    logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+                        level=(logging.DEBUG if args.verbose else logging.INFO))
+
+    with open(args.config, 'r') as f:
+        config = yaml.safe_load(f)
+
+    if not isinstance(config, dict):
+        raise TypeError('invalid config')
+
+    results = Results()
+    max_threads = math.inf if args.threads_limit == 0 else args.threads_limit
+    active_threads = 1
+
+    def get_active_threads():
+        n = active_threads
+        if workers:
+            n += workers[0].concurrency
+        return n
+
+    workers = []
+    for name, data in config.items():
+        w = Worker(name, data['host'], data['opened'],
+                   concurrency=int(data['concurrency']) if 'concurrency' in data else args.concurrency,
+                   timeout=int(data['timeout']) if 'timeout' in data else args.timeout)
+        workers.append(w)
+
+    current_workers = []
+    while workers:
+        w = workers.pop(0)
+        active_threads += w.concurrency+1
+
+        current_workers.append(w)
+        w.start()
+
+        while current_workers and get_active_threads() >= max_threads:
+            for cw in current_workers:
+                cw.join(timeout=0.1)
+                if not cw.is_alive():
+                    results.add(cw)
+                    current_workers.remove(cw)
+                    active_threads -= cw.concurrency+1
+
+    for cw in current_workers:
+        cw.join()
+        results.add(cw)
+
+    if results.has_warnings() and not args.no_telegram:
+        results.notify()
+
+
+if __name__ == '__main__':
+    main()