Add TCP support

README.md

@@ -1,20 +0,0 @@
-# velping
-a pinger. it pings. this is made specifically for my needs, so you can open an issue for a feature request, but i likely won't accept it unless it's something i want.
-
-## configuration options
-### frontend
-- `name`: what will show as the header on the landing page.
-- `port`: what port the frontend will run on.
-- `pings`: how many pings to show on the frontend
-
-### pinging
-- `allow_empty_responses`: this does NOT mean HTTP 204, but rather if the connection being suddenly closed after a request should be accepted as online. this option only exists because i run a minecraft server.
-- `seconds_between_ping`: how often services should be pinged, in seconds.
-
-### ntfy
-- `enabled`: whether or not to enable ntfy.
-- `server`: the server to send notifications to.
-- `topic`: the ntfy topic.
-
-### services
-you can add services here, with the name of the key being the name of the service and the value being the URL to ping.

config-example.toml

@@ -4,7 +4,6 @@ name = "velping"
 pings = 25
 
 [pinging]
-allow_empty_responses = false
 seconds_between_ping = 60
 
 [ntfy]
@@ -12,5 +11,16 @@ enabled = false
 server = "ntfy.sh"
 topic = "my_velping_instance"
 
-[services]
+[services.google]
-Google = "https://google.com/"
+name = "Google"
+type = "http"
+address = "https://google.com/"
+timeout = 5
+
+[services.iana-whois]
+name = "IANA WHOIS"
+type = "tcp"
+port = 43
+hostname = "whois.iana.org"
+timeout = 5
+ip_version = 4

index.py

@@ -3,6 +3,7 @@ import threading
 import requests
 import tomllib
 import logging
+import socket
 import time
 
 with open("config.toml", "rb") as f:
@@ -14,59 +15,140 @@ logging.disable(logging.CRITICAL)
 uptime = {}
 
 
-def http_ping(service):
+def handle_service_status(
+	service_name: str,
+	down: bool = False,
+	error: str | None = None,
+):
+	if (
+		uptime[service_name][config["frontend"]["pings"] - 1] != "O"
+		and config["ntfy"]["enabled"]
+		and down
+	):
+		requests.post(
+			url=f"https://{config['ntfy']['server']}/{config['ntfy']['topic']}",
+			data=f"Service {service_name} is offline",
+		)
+	elif (
+		uptime[service_name][config["frontend"]["pings"] - 1] != "I"
+		and config["ntfy"]["enabled"]
+		and not down
+	):
+		requests.post(
+			url=f"https://{config['ntfy']['server']}/{config['ntfy']['topic']}",
+			data=f"Service {service_name} is online",
+		)
+
+	uptime[service_name].pop(0)
+	if down:
+		uptime[service_name].append("O")
+		print(f"[E] An error happened while pinging for {service_name}: {error}")
+	else:
+		uptime[service_name].append("I")
+		print(f"[I] Pinging {service_name} succeeded!")
+
+
+def tcp_ping(service):
+	ip = service["ip_version"]
+	name = service["name"]
+	host = service["hostname"]
+	port = service["port"]
+	timeout = service["timeout"]
+
+	if ip == 4:
+		socket_type = socket.AF_INET
+	elif ip == 6:
+		socket_type = socket.AF_INET6
+	else:
+		print(f"[E] IP version for {name} is not 4 or 6")
+		return
+
 	while True:
-		print(f"[I] Pinging {service}")
+		print(f"[I] Pinging {name}")
+
 		try:
-			resp = requests.get(config["services"][service])
+			with socket.socket(family=socket_type, type=socket.SOCK_STREAM) as sock:
-			resp.raise_for_status()
+				sock.settimeout(timeout)
+				sock.connect((host, port))
 
-			if (
+			handle_service_status(
-				uptime[service][config["frontend"]["pings"] - 1] != "I"
+				service_name=name,
-				and config["ntfy"]["enabled"]
+				down=False,
-			):
+			)
-				requests.post(
-					f"https://{config['ntfy']['server']}/{config['ntfy']['topic']}",
-					data=f"Service {service} is online",
-				)
 
-			uptime[service].pop(0)
+		except socket.timeout:
-			uptime[service].append("I")
+			handle_service_status(
-			print(f"[I] Pinging {service} worked!")
+				service_name=name,
-		except Exception as e:
+				down=True,
-			if (
+				error=f"[E] Connection to {host}:{port} timed out",
-				"Remote end closed connection without response" in str(e)
+			)
-				and config["pinging"]["allow_empty_responses"]
-			):
-				uptime[service].pop(0)
-				uptime[service].append("I")
-				print(f"[I] Pinging {service} worked!")
-			else:
-				if (
-					uptime[service][config["frontend"]["pings"] - 1] != "O"
-					and config["ntfy"]["enabled"]
-				):
-					requests.post(
-						f"https://{config['ntfy']['server']}/{config['ntfy']['topic']}",
-						data=f"Service {service} is offline",
-					)
 
-				uptime[service].pop(0)
+		except Exception as err:
-				uptime[service].append("O")
+			handle_service_status(
-				print(f"[E] An error happened while pinging for {service}: {e}")
+				service_name=name,
+				down=True,
+				error=f"[E] Connection error: {err}",
+			)
 
 		time.sleep(config["pinging"]["seconds_between_ping"])
 
 
-for service in config["services"]:
+def http_ping(service):
-	uptime[service] = []
+	name = service["name"]
-	for _ in range(config["frontend"]["pings"]):
+	address = service["address"]
-		uptime[service].append("?")
+	timeout = service["timeout"]
 
-	threading.Thread(
+	while True:
-		target=http_ping,
+		print(f"[I] Pinging {service['name']}")
-		args=(service,),
+		try:
-	).start()
+			resp = requests.get(
+				url=address,
+				timeout=timeout,
+			)
+			resp.raise_for_status()
+
+			handle_service_status(
+				service_name=name,
+				down=False,
+			)
+		except Exception as e:
+			handle_service_status(
+				service_name=name,
+				down=True,
+				error=str(e),
+			)
+
+		time.sleep(config["pinging"]["seconds_between_ping"])
+
+
+for service_name in config["services"]:
+	service = config["services"][service_name]
+	if "type" in service and "name" in service and "timeout" in service:
+		uptime[service["name"]] = []
+		for _ in range(config["frontend"]["pings"]):
+			uptime[service["name"]].append("?")
+
+		if service["type"] == "http" and "address" in service:
+			threading.Thread(
+				target=http_ping,
+				args=(service,),
+			).start()
+		elif (
+			service["type"] == "tcp"
+			and "port" in service
+			and "hostname" in service
+			and "ip_version" in service
+		):
+			threading.Thread(
+				target=tcp_ping,
+				args=(service,),
+			).start()
+		else:
+			print(
+				f"[W] Service {service['name']} is missing required fields, ignoring."
+			)
+	else:
+		print("[W] Invalid service entry found, ignoring.")
 
 
 @app.errorhandler(404)

templates/index.xht

@@ -12,11 +12,11 @@
 <body>
 	<h2>{{ config['frontend']['name'] }}</h2>
 
-	{% for service in services %}
+	{% for service_name in services %}
 		<div class="service">
-			<p><b>{{ service }}</b></p>
+			<p><b>{{ services[service_name]['name'] }}</b></p>
 			<div class="status">
-				{% for ping in uptime[service] %}
+				{% for ping in uptime[services[service_name]['name']] %}
 					{% if ping == "I" %}
 						<p class="green">█</p>
 					{% elif ping == "O" %}