This commit is contained in:
Mateusz Gruszczyński 2025-03-06 23:01:21 +01:00
parent 556d74816e
commit 86d4bd1904
3 changed files with 43 additions and 9 deletions

1
.gitignore vendored
View File

@ -2,3 +2,4 @@ __pycache__
data/ data/
instance/ instance/
venv/ venv/
config.ini

51
app.py
View File

@ -13,10 +13,24 @@ from croniter import croniter
app = Flask(__name__) app = Flask(__name__)
sslify = SSLify(app) sslify = SSLify(app)
# Konfiguracja tokenu autoryzacyjnego (przyjmujemy pre-shared secret) # Ścieżka do pliku z tokenem:
API_TOKEN = os.environ.get("HOSTS_DAEMON_API_TOKEN", "superSecretTokenABC123") TOKEN_FILE_PATH = "/opt/hosts_daemon/token"
# Konfiguracja logowania logi zapisywane do pliku /opt/hosts_daemon/logs/daemon.log def read_token_from_file(path):
"""Odczytuje token z pliku i zwraca jego zawartość (strip),
albo None, jeśli plik nie istnieje lub jest pusty."""
if os.path.isfile(path):
try:
with open(path, 'r') as f:
content = f.read().strip()
if content:
return content
except Exception as e:
# Możesz zalogować błąd, jeśli np. plik jest niewidoczny/brak uprawnień
logger.error(f"Nie udało się odczytać pliku tokenu: {str(e)}")
return None
# LOGOWANIE
LOG_DIR = "/opt/hosts_daemon/logs" LOG_DIR = "/opt/hosts_daemon/logs"
os.makedirs(LOG_DIR, exist_ok=True) os.makedirs(LOG_DIR, exist_ok=True)
LOG_FILE = os.path.join(LOG_DIR, "daemon.log") LOG_FILE = os.path.join(LOG_DIR, "daemon.log")
@ -30,6 +44,24 @@ logging.basicConfig(
) )
logger = logging.getLogger("hosts_daemon") logger = logging.getLogger("hosts_daemon")
# 1) Najpierw próbujemy odczytać token z pliku
file_token = read_token_from_file(TOKEN_FILE_PATH)
# 2) Jeśli w pliku nie ma tokenu, sprawdzamy zmienną środowiskową
if file_token:
# Użyjemy tokenu z pliku
API_TOKEN = file_token
logger.info("API_TOKEN wczytany z pliku.")
else:
# Fallback: odczytujemy z ENV albo dajemy domyślny
env_token = os.environ.get("HOSTS_DAEMON_API_TOKEN")
if env_token:
API_TOKEN = env_token
logger.info("API_TOKEN wczytany ze zmiennej środowiskowej.")
else:
API_TOKEN = "superSecretTokenABC123"
logger.info("API_TOKEN ustawiony na wartość domyślną: superSecretTokenABC123")
# Globalne metryki # Globalne metryki
metrics = { metrics = {
"total_requests": 0, "total_requests": 0,
@ -46,7 +78,9 @@ metrics = {
def require_auth(): def require_auth():
"""Wymusza autoryzację przy pomocy nagłówka Authorization, który powinien zawierać API_TOKEN.""" """Wymusza autoryzację przy pomocy nagłówka Authorization, który powinien zawierać API_TOKEN."""
token = request.headers.get("Authorization") token = request.headers.get("Authorization")
logger.info(f"require_auth() -> Nagłówek Authorization: {token}")
if token != API_TOKEN: if token != API_TOKEN:
logger.warning("Nieprawidłowy token w nagłówku Authorization. Oczekiwano innego ciągu znaków.")
abort(401, description="Unauthorized") abort(401, description="Unauthorized")
def validate_hosts_syntax(hosts_content): def validate_hosts_syntax(hosts_content):
@ -59,7 +93,8 @@ def validate_hosts_syntax(hosts_content):
(ograniczone do nowej treści, nie sprawdza starych plików). (ograniczone do nowej treści, nie sprawdza starych plików).
Zwraca None, jeśli OK, w przeciwnym razie string z opisem błędu. Zwraca None, jeśli OK, w przeciwnym razie string z opisem błędu.
""" """
# Słownik do wykrywania duplikatów: (ip, hostname) -> bool import ipaddress
seen = {} seen = {}
lines = hosts_content.splitlines() lines = hosts_content.splitlines()
for i, line in enumerate(lines, start=1): for i, line in enumerate(lines, start=1):
@ -75,16 +110,12 @@ def validate_hosts_syntax(hosts_content):
ip_addr = parts[0] ip_addr = parts[0]
hostnames = parts[1:] hostnames = parts[1:]
# Prosta weryfikacja IP - w Pythonie można użyć ipaddress, ale zrobimy skrótowo: # Prosta weryfikacja IP
import ipaddress
try: try:
_ = ipaddress.ip_address(ip_addr) _ = ipaddress.ip_address(ip_addr)
except ValueError: except ValueError:
return f"Linia {i}: '{ip_addr}' nie jest poprawnym adresem IP" return f"Linia {i}: '{ip_addr}' nie jest poprawnym adresem IP"
if not hostnames:
return f"Linia {i}: brak hostnamów."
for hn in hostnames: for hn in hostnames:
key = (ip_addr, hn) key = (ip_addr, hn)
if key in seen: if key in seen:
@ -222,6 +253,8 @@ def system_info():
return jsonify(info), 200 return jsonify(info), 200
if __name__ == '__main__': if __name__ == '__main__':
# Możesz zalogować tylko informację, że startujesz:
logger.info("Uruchamiam hosts_daemon token wczytany, nasłuch na porcie 8000 (HTTPS).")
app.run( app.run(
host='0.0.0.0', host='0.0.0.0',
port=8000, port=8000,

0
daemon_token.txt Normal file
View File