feat: add helper scripts for infrastructure management and troubleshooting

This commit is contained in:
2026-11-06 15:24:05 -05:00
parent aea8608f3d
commit 1965d37687
48 changed files with 1399 additions and 39 deletions

View File

@@ -0,0 +1,52 @@
import paramiko
import time
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def apply_and_update(ip, name, manifest_path):
print(f"\n=======================================================")
print(f"Applying to {name} ({ip})...")
with open(manifest_path, 'r', encoding='utf-8') as f:
content = f.read()
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ip, username='root', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl apply -f -")
stdin.write(content)
stdin.channel.shutdown_write()
print("Apply output:", stdout.read().decode())
err = stderr.read().decode()
if err:
print("Apply stderr:", err)
print(f"Waiting for Pi-hole deployment rollout on {name}...")
stdin, stdout, stderr = ssh.exec_command("k3s kubectl rollout status deployment/pihole -n dns --timeout=60s")
print(stdout.read().decode())
# Get new pod name
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n dns -l app=pihole -o jsonpath='{.items[0].metadata.name}'")
pod_name = stdout.read().decode().strip()
print(f"New Pi-hole pod: {pod_name}")
# Set password to snarfnetadmin
print("Setting Pi-hole password to 'snarfnetadmin'...")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n dns {pod_name} -- pihole setpassword snarfnetadmin")
out = stdout.read().decode()
err = stderr.read().decode()
print("Set password output:", out)
if err:
print("Set password stderr:", err)
ssh.close()
aero_manifest = r"C:\Users\Snarf\OneDrive\Desktop\code\SnarfCode\homelab-gitops-repo\manifests\aerodrome\dns\dns-stack.yaml"
space_manifest = r"C:\Users\Snarf\OneDrive\Desktop\code\SnarfCode\homelab-gitops-repo\manifests\spaceport\dns\dns-stack.yaml"
apply_and_update("140.44.4.11", "Aerodrome", aero_manifest)
apply_and_update("140.44.4.41", "Spaceport", space_manifest)
print("\nDone!")

140
scripts/apply_npm_fixes.py Normal file
View File

@@ -0,0 +1,140 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect("140.44.4.71", username="root", key_filename=key_path, timeout=10)
print("1. Creating backup of database and configs...")
backup_cmd = """
cp /var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite /var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite.bak.$(date +%s)
"""
stdin, stdout, stderr = c.exec_command(backup_cmd)
stdout.channel.recv_exit_status()
print("Backup complete.")
print("2. Updating database.sqlite for hosts 45, 41, and 37...")
py_db = """
import sqlite3, datetime
conn = sqlite3.connect("/var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite")
cur = conn.cursor()
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
updates = [
(45, "140.44.4.34", 443, "https"), # Longhorn -> Aerodrome Traefik
(41, "140.44.4.62", 443, "https"), # Semaphore -> Spaceport Traefik
(37, "140.44.4.34", 443, "https"), # Uptime Kuma -> Aerodrome Traefik
]
for pid, host, port, scheme in updates:
cur.execute('''UPDATE proxy_host
SET forward_host = ?, forward_port = ?, forward_scheme = ?,
allow_websocket_upgrade = 1, http2_support = 1, modified_on = ?
WHERE id = ?''', (host, port, scheme, now, pid))
print(f"DB updated: Host {pid} -> {scheme}://{host}:{port}")
conn.commit()
conn.close()
"""
stdin, stdout, stderr = c.exec_command(f"python3 -c '{py_db}'")
print(stdout.read().decode())
err = stderr.read().decode()
if err: print("DB err:", err)
print("3. Generating updated nginx configuration files...")
def generate_conf(pid, domains, cert_id, host, port, scheme):
return f"""# ------------------------------------------------------------
# {domains}
# ------------------------------------------------------------
map $scheme $hsts_header {{
https "max-age=63072000; preload";
}}
server {{
set $forward_scheme {scheme};
set $server "{host}";
set $port {port};
listen 80;
listen [::]:80;
listen 443 ssl;
listen [::]:443 ssl;
server_name {domains};
http2 on;
# Let's Encrypt SSL
include conf.d/include/letsencrypt-acme-challenge.conf;
include conf.d/include/ssl-cache.conf;
include conf.d/include/ssl-ciphers.conf;
ssl_certificate /etc/letsencrypt/live/npm-{cert_id}/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/npm-{cert_id}/privkey.pem;
# Asset Caching
include conf.d/include/assets.conf;
# Block Exploits
include conf.d/include/block-exploits.conf;
# Force SSL
set $trust_forwarded_proto "F";
include conf.d/include/force-ssl.conf;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $http_connection;
proxy_http_version 1.1;
access_log /data/logs/proxy-host-{pid}_access.log proxy;
error_log /data/logs/proxy-host-{pid}_error.log warn;
location / {{
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $http_connection;
proxy_http_version 1.1;
# Proxy!
include conf.d/include/proxy.conf;
}}
# Custom
include /data/nginx/custom/server_proxy[.]conf;
}}
"""
configs = [
(45, "longhorn.aerodrome.snarfnet.net", 37, "140.44.4.34", 443, "https"),
(41, "semaphore.snarfnet.net", 33, "140.44.4.62", 443, "https"),
(37, "uptime.snarfnet.net", 27, "140.44.4.34", 443, "https"),
]
for pid, dom, cert_id, host, port, scheme in configs:
cfg = generate_conf(pid, dom, cert_id, host, port, scheme)
path = f"/var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/nginx/proxy_host/{pid}.conf"
cmd = f"cat << 'EOF' > {path}\n{cfg}\nEOF\n"
stdin, stdout, stderr = c.exec_command(cmd)
stdout.channel.recv_exit_status()
print(f"Wrote config: {path}")
print("4. Testing Nginx config inside container...")
stdin, stdout, stderr = c.exec_command("docker exec $(docker ps -q -f name=nginx-proxy-manager_app | head -n1) nginx -t")
out = stdout.read().decode()
err = stderr.read().decode()
print("nginx -t stdout:", out)
print("nginx -t stderr:", err)
if "successful" in out or "successful" in err:
print("5. Reloading Nginx via Swarm service update...")
stdin, stdout, stderr = c.exec_command("docker service update --force nginx-proxy-manager_app")
print(stdout.read().decode())
print("Reload completed.")
else:
print("ERROR: Nginx config test failed!")
c.close()

View File

@@ -0,0 +1,72 @@
import paramiko
import urllib.request
import ssl
import json
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
print("1. Updating Authentik Provider 10 redirect_uris...")
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
token = "SnarfNetPermanentAutomationToken2026"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
provider_update = {
"redirect_uris": [
{"matching_mode": "strict", "url": "https://semaphore.snarfnet.net/auth/oidc/authentik/redirect"},
{"matching_mode": "strict", "url": "https://semaphore.snarfnet.net/api/auth/oidc/authentik/redirect/"},
{"matching_mode": "regex", "url": "https://semaphore\\.snarfnet\\.net/.*"}
]
}
req = urllib.request.Request(
"https://authentik.snarfnet.net/api/v3/providers/oauth2/10/",
headers=headers,
data=json.dumps(provider_update).encode('utf-8'),
method="PATCH"
)
try:
with urllib.request.urlopen(req, context=ctx, timeout=8) as resp:
res = json.loads(resp.read().decode())
print(" ✓ Authentik Provider 10 updated successfully:", res.get('redirect_uris'))
except Exception as e:
print(" ✗ Failed to update Authentik Provider 10:", e)
print("\n2. Applying updated Semaphore manifest to Spaceport (140.44.4.41)...")
manifest_path = r"C:\Users\Snarf\OneDrive\Desktop\code\SnarfCode\homelab-gitops-repo\manifests\spaceport\semaphore\semaphore.yaml"
with open(manifest_path, 'r', encoding='utf-8') as f:
content = f.read()
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl apply -f -")
stdin.write(content)
stdin.channel.shutdown_write()
print("Apply output:", stdout.read().decode())
err = stderr.read().decode()
if err: print("Apply stderr:", err)
print("3. Rolling out Semaphore deployment...")
stdin, stdout, stderr = ssh.exec_command("k3s kubectl rollout restart deployment/semaphore -n infrastructure; k3s kubectl rollout status deployment/semaphore -n infrastructure --timeout=90s")
print(stdout.read().decode())
# Check new pod env
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n infrastructure -l app.kubernetes.io/name=semaphore -o jsonpath='{.items[0].metadata.name}'")
new_pod = stdout.read().decode().strip()
print(f"New pod: {new_pod}")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {new_pod} -- env | grep SEMAPHORE_OIDC_PROVIDERS")
print("New env:", stdout.read().decode())
ssh.close()
print("Done!")

View File

@@ -0,0 +1,26 @@
import urllib.request
import ssl
import json
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
token = "SnarfNetPermanentAutomationToken2026"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
req = urllib.request.Request("https://authentik.snarfnet.net/api/v3/providers/oauth2/", headers=headers)
with urllib.request.urlopen(req, context=ctx) as resp:
data = json.loads(resp.read().decode())
for p in data['results']:
if 'semaphore' in p['name'].lower():
print(f"Provider ID: {p['pk']}, Name: {p['name']}")
print(f"Redirect URIs: {p.get('redirect_uris')}")
print(f"Client ID: {p.get('client_id')}")

View File

@@ -0,0 +1,20 @@
import urllib.request
import ssl
import json
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
token = "SnarfNetPermanentAutomationToken2026"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
req = urllib.request.Request("https://authentik.snarfnet.net/api/v3/core/users/?page_size=20", headers=headers)
with urllib.request.urlopen(req, context=ctx) as resp:
data = json.loads(resp.read().decode())
for u in data['results']:
print(f"User: {u['username']}, Email: {u.get('email')}, Name: {u.get('name')}")

21
scripts/check_cert_san.py Normal file
View File

@@ -0,0 +1,21 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect("140.44.4.71", username="root", key_filename=key_path, timeout=10)
def check_cert(num):
cmd = f"docker exec $(docker ps -q -f name=nginx-proxy-manager_app | head -n1) openssl x509 -in /etc/letsencrypt/live/{num}/fullchain.pem -noout -text | grep -A 2 'Subject Alternative Name'"
stdin, stdout, stderr = c.exec_command(cmd)
print(f"Cert {num}:", stdout.read().decode().strip())
check_cert("npm-1")
check_cert("npm-33")
check_cert("npm-37")
c.close()

View File

@@ -1,17 +1,14 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('140.44.4.11', username='root', key_filename=r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet', timeout=10)
ssh.connect('140.44.4.41', username='root', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command('k3s kubectl exec -n dns deployment/pihole -- pihole-FTL --config webserver.api.password')
print("Current config webserver.api.password:")
print(stdout.read().decode('utf-8', errors='ignore'))
# Check pihole-FTL --config --help or options
stdin, stdout, stderr = ssh.exec_command('k3s kubectl exec -n dns deployment/pihole -- pihole-FTL --config 2>&1 || true')
print("pihole-FTL --config:")
print(stdout.read().decode('utf-8', errors='ignore')[:1500])
stdin, stdout, stderr = ssh.exec_command("k3s kubectl exec -n dns pihole-88584bd7-d5vq7 -- sed -n '57,85p' /usr/local/bin/pihole")
print(stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,15 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get deployment -n infrastructure semaphore -o jsonpath='{.spec.template.spec.containers[0].env[?(@.name==\"SEMAPHORE_OIDC_PROVIDERS\")].value}'")
print("Deployment live env:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,33 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def run_ssh(host, user, cmd):
print(f"=== SSH {user}@{host}: {cmd} ===")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(host, username=user, key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
if out:
print(out)
if err:
print("STDERR:", err)
except Exception as e:
print(f"Error connecting to {host}: {e}")
finally:
ssh.close()
# Aerodrome Longhorn Ingress
run_ssh('140.44.4.11', 'root', 'k3s kubectl get ingress -A | grep -i longhorn')
run_ssh('140.44.4.11', 'root', 'k3s kubectl get ingress -n longhorn-system -o yaml')
# Spaceport Semaphore Ingress & Deployment
run_ssh('140.44.4.41', 'root', 'k3s kubectl get ingress -A | grep -i semaphore')
run_ssh('140.44.4.41', 'root', 'k3s kubectl get deployment -A | grep -i semaphore')
run_ssh('140.44.4.41', 'root', 'k3s kubectl get deployment -n infrastructure semaphore -o jsonpath="{.spec.template.spec.containers[0].env}"')

24
scripts/check_npm_37.py Normal file
View File

@@ -0,0 +1,24 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect("140.44.4.71", username="root", key_filename=key_path, timeout=10)
py_check = """
import sqlite3
conn = sqlite3.connect("/var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite")
cur = conn.cursor()
for row in cur.execute("SELECT id, domain_names, forward_host, forward_port, forward_scheme, certificate_id FROM proxy_host WHERE id = 37"):
print("Row:", row)
conn.close()
"""
stdin, stdout, stderr = c.exec_command(f"python3 -c '{py_check}'")
print(stdout.read().decode())
c.close()

View File

@@ -0,0 +1,29 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect("140.44.4.71", username="root", key_filename=key_path, timeout=10)
py_check = """
import sqlite3
conn = sqlite3.connect("/var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite")
cur = conn.cursor()
for row in cur.execute("SELECT id, domain_names, forward_host, forward_port, forward_scheme, certificate_id FROM proxy_host WHERE id IN (45, 41)"):
print("Row:", row)
conn.close()
"""
stdin, stdout, stderr = c.exec_command(f"python3 -c '{py_check}'")
print(stdout.read().decode())
err = stderr.read().decode()
if err: print("Err:", err)
stdin, stdout, stderr = c.exec_command("docker exec $(docker ps -q -f name=nginx-proxy-manager_app | head -n1) ls -la /etc/letsencrypt/live/")
print("Certs:\n", stdout.read().decode())
c.close()

18
scripts/check_npm_host.py Normal file
View File

@@ -0,0 +1,18 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('140.44.4.71', username='snarf', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("sudo docker inspect $(sudo docker ps -q -f name=nginx-proxy-manager) | grep -A 5 -B 5 Mounts")
print("Docker mounts:\n", stdout.read().decode())
stdin, stdout, stderr = ssh.exec_command("sudo which sqlite3; python3 -c 'import sqlite3; print(\"python sqlite3 ok\")'")
print("Host tools:\n", stdout.read().decode())
ssh.close()

24
scripts/check_npm_nas.py Normal file
View File

@@ -0,0 +1,24 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect("140.44.4.71", username="root", key_filename=key_path, timeout=10)
py_check = """
import sqlite3
conn = sqlite3.connect("/var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite")
cur = conn.cursor()
for row in cur.execute("SELECT id, domain_names, forward_host, forward_port, forward_scheme, certificate_id FROM proxy_host WHERE id IN (11, 28)"):
print("Row:", row)
conn.close()
"""
stdin, stdout, stderr = c.exec_command(f"python3 -c '{py_check}'")
print(stdout.read().decode())
c.close()

View File

@@ -0,0 +1,30 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def check_cluster(ip, name):
print(f"\n--- Checking Pi-hole on {name} ({ip}) ---")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ip, username='root', key_filename=key_path, timeout=10)
# Check Pi-hole pod name
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n dns -l app=pihole -o jsonpath='{.items[0].metadata.name}'")
pod_name = stdout.read().decode().strip()
print(f"Pod name: {pod_name}")
# Check password in setupVars or FTL
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n dns {pod_name} -- grep -i pwhash /etc/pihole/pihole-FTL.conf /etc/pihole/setupVars.conf")
print("Password hash check:\n", stdout.read().decode())
# Test curl to localhost:80/admin/
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n dns {pod_name} -- curl -s -I http://localhost/admin/")
print("Local admin response:\n", stdout.read().decode())
ssh.close()
check_cluster('140.44.4.11', 'Aerodrome')
check_cluster('140.44.4.41', 'Spaceport')

View File

@@ -0,0 +1,29 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def run_ssh(host, cmd):
print(f"=== SSH {host}: {cmd} ===")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(host, username='snarf', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
if out:
print(out)
if err:
print("STDERR:", err)
except Exception as e:
print(f"Error connecting to {host}: {e}")
finally:
ssh.close()
# Check Spaceport (140.44.4.41) Pi-hole pod, service, secret
run_ssh('140.44.4.41', "sudo kubectl get pods -n dns -o wide; sudo kubectl get svc -n dns -o wide")
run_ssh('140.44.4.41', "sudo kubectl get secret -n dns -o yaml | grep -E 'name:|WEBPASSWORD|password'")

View File

@@ -0,0 +1,17 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
pod = "semaphore-98778cd46-sjs7h"
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- find / -name '*app*.js' -o -name '*oidc*.js' -o -name 'index.html' 2>/dev/null")
print("Web assets:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,19 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n infrastructure -l app.kubernetes.io/name=semaphore -o jsonpath='{.items[0].metadata.name}'")
pod = stdout.read().decode().strip()
print(f"Active pod: {pod}")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl logs -n infrastructure {pod} --tail=100")
print("Logs:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,20 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
pod = "semaphore-98778cd46-sjs7h"
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl get pod -n infrastructure {pod} -o jsonpath='{{.spec.containers[0].volumeMounts}}'")
print("Volume mounts:\n", stdout.read().decode())
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl get pod -n infrastructure {pod} -o jsonpath='{{.spec.volumes}}'")
print("Volumes:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,16 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
pod = "semaphore-98778cd46-sjs7h"
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- cat /usr/local/bin/server-wrapper")
print("server-wrapper:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,19 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
pod = "semaphore-5c96655c49-xv456"
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- /usr/local/bin/semaphore user list --config /etc/semaphore/config.json")
print("Semaphore users:\n", stdout.read().decode())
err = stderr.read().decode()
if err: print("Error:\n", err)
ssh.close()

View File

@@ -0,0 +1,14 @@
import urllib.request
import ssl
import re
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
url = "https://semaphore.snarfnet.net/js/app.905c93b7.js"
content = urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}), context=ctx).read().decode('utf-8')
routes = re.findall(r'path:\s*["\']([^"\']+)["\']', content)
print("Routes in Vue app:", set(routes))

17
scripts/find_oidc_js.py Normal file
View File

@@ -0,0 +1,17 @@
import urllib.request
import ssl
import re
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
url = "https://semaphore.snarfnet.net/js/app.905c93b7.js"
content = urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}), context=ctx).read().decode('utf-8')
for m in re.finditer(r'oidc', content, re.IGNORECASE):
start = max(0, m.start() - 150)
end = min(len(content), m.end() + 150)
print("Match:", content[start:end])
print("-" * 50)

View File

@@ -0,0 +1,18 @@
import urllib.request
import ssl
import re
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
url = "https://semaphore.snarfnet.net/js/app.905c93b7.js"
content = urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}), context=ctx).read().decode('utf-8')
for m in re.finditer(r'/auth/oidc/[^"\']+', content):
print("Route match:", m.group(0))
# Also search for router definitions
for m in re.finditer(r'path:\s*["\'][^"\']*oidc[^"\']*["\']', content, re.IGNORECASE):
print("Router path:", m.group(0))

View File

@@ -0,0 +1,38 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
with open(r"C:\Users\Snarf\OneDrive\Desktop\code\SnarfCode\homelab-gitops-repo\manifests\spaceport\semaphore\semaphore.yaml", "r", encoding="utf-8") as f:
manifest_content = f.read()
print("Applying with --force-conflicts=true...")
stdin, stdout, stderr = ssh.exec_command("k3s kubectl apply -f - --server-side=true --force-conflicts=true")
stdin.write(manifest_content)
stdin.channel.shutdown_write()
print("Apply stdout:", stdout.read().decode())
print("Apply stderr:", stderr.read().decode())
print("Refreshing ArgoCD Application...")
# Check if argocd CLI or patch annotation
stdin, stdout, stderr = ssh.exec_command("k3s kubectl annotate application spaceport-semaphore -n infrastructure argocd.argoproj.io/refresh=hard --overwrite")
print(stdout.read().decode())
print("Waiting for deployment rollout...")
stdin, stdout, stderr = ssh.exec_command("k3s kubectl rollout status deployment/semaphore -n infrastructure --timeout=60s")
print(stdout.read().decode())
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n infrastructure -l app.kubernetes.io/name=semaphore -o jsonpath='{.items[0].metadata.name}'")
pod = stdout.read().decode().strip()
print(f"Active pod: {pod}")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- env | grep SEMAPHORE_OIDC_PROVIDERS")
print("Live env:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,30 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('140.44.4.71', username='snarf', key_filename=key_path, timeout=10)
def run_cmd(cmd):
stdin, stdout, stderr = ssh.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
return out, err
out, _ = run_cmd("sudo docker exec $(sudo docker ps -q -f name=nginx-proxy-manager) sqlite3 /data/database.sqlite 'SELECT id, domain_names, forward_scheme, forward_host, forward_port FROM proxy_host WHERE id IN (45, 41);'")
print("SQLite proxy_host entries:")
print(out)
out, _ = run_cmd("sudo docker exec $(sudo docker ps -q -f name=nginx-proxy-manager) cat /data/nginx/proxy_host/45.conf")
print("=== 45.conf ===")
print(out)
out, _ = run_cmd("sudo docker exec $(sudo docker ps -q -f name=nginx-proxy-manager) cat /data/nginx/proxy_host/41.conf")
print("=== 41.conf ===")
print(out)
ssh.close()

View File

@@ -0,0 +1,29 @@
import urllib.request
import ssl
import re
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request("https://semaphore.snarfnet.net/", headers={"User-Agent": "Mozilla/5.0"})
html = urllib.request.urlopen(req, context=ctx).read().decode('utf-8')
print("HTML:", html[:500])
# Find script tags
scripts = re.findall(r'src="([^"]+\.js)"', html)
print("Scripts found:", scripts)
for s in scripts:
s_url = f"https://semaphore.snarfnet.net{s}" if s.startswith('/') else f"https://semaphore.snarfnet.net/{s}"
s_req = urllib.request.Request(s_url, headers={"User-Agent": "Mozilla/5.0"})
content = urllib.request.urlopen(s_req, context=ctx).read().decode('utf-8', errors='replace')
if 'oidc' in content.lower() or 'authentik' in content.lower():
print(f"\n--- Match in {s} ---")
matches = re.findall(r'.{0,100}(?:oidc|login|redirect).{0,100}', content, re.IGNORECASE)
for m in matches[:10]:
print(" ...", m.strip(), "...")

View File

@@ -0,0 +1,22 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('140.44.4.41', username='root', key_filename=key_path, timeout=10)
pod_name = "semaphore-786cf4fc8d-vfvq8"
def test(path):
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod_name} -- curl -s -I 'http://localhost:3000{path}'")
print(f"{path} -> {stdout.read().decode().strip()}")
test("/api/auth/oidc/authentik/login")
test("/api/auth/oidc/authentik/redirect")
test("/auth/oidc/authentik/auth/login")
test("/auth/oidc/authentik/redirect")
ssh.close()

View File

@@ -0,0 +1,67 @@
import paramiko
import sys
import json
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
oidc_config = {
"authentik": {
"display_name": "Sign in with Authentik",
"provider_url": "https://authentik.snarfnet.net/application/o/semaphore/",
"client_id": "semaphore-2e50dbdf",
"client_secret": "V4uazPXN9toUn5Jf2DfsRIFrygf1T0Vcltga-APrzTRjWFN0DIIqNw",
"redirect_url": "https://semaphore.snarfnet.net/auth/oidc/authentik/redirect",
"username_claim": "preferred_username",
"name_claim": "name",
"email_claim": "email",
"scopes": ["openid", "profile", "email", "groups"]
}
}
val = json.dumps(oidc_config)
# Use patch json
patch = {
"spec": {
"template": {
"spec": {
"containers": [
{
"name": "semaphore",
"env": [
{
"name": "SEMAPHORE_OIDC_PROVIDERS",
"value": val
}
]
}
]
}
}
}
}
patch_str = json.dumps(patch)
print("Patching deployment/semaphore...")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl patch deployment semaphore -n infrastructure -p '{patch_str}'")
print(stdout.read().decode())
err = stderr.read().decode()
if err: print("Patch stderr:", err)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl rollout status deployment/semaphore -n infrastructure --timeout=60s")
print(stdout.read().decode())
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n infrastructure -l app.kubernetes.io/name=semaphore -o jsonpath='{.items[0].metadata.name}'")
new_pod = stdout.read().decode().strip()
print(f"Active pod: {new_pod}")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {new_pod} -- env | grep SEMAPHORE_OIDC_PROVIDERS")
print("Active pod env:\n", stdout.read().decode())
ssh.close()

28
scripts/set_pihole_pw.py Normal file
View File

@@ -0,0 +1,28 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def set_pw(ip, name):
print(f"\n--- Setting Pi-hole password on {name} ({ip}) ---")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ip, username='root', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n dns -l app=pihole -o jsonpath='{.items[0].metadata.name}'")
pod_name = stdout.read().decode().strip()
# Try setting password to snarfnetadmin
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n dns {pod_name} -- pihole setpassword snarfnetadmin")
out = stdout.read().decode()
err = stderr.read().decode()
print("setpassword output:", out)
if err:
print("setpassword err:", err)
ssh.close()
set_pw('140.44.4.11', 'Aerodrome')
set_pw('140.44.4.41', 'Spaceport')

View File

@@ -0,0 +1,44 @@
import paramiko
import sys
import time
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
print("1. Triggering ArgoCD sync on spaceport-semaphore...")
# If argocd CLI or kubectl apply
stdin, stdout, stderr = ssh.exec_command("k3s kubectl apply -f C:/Users/Snarf/OneDrive/Desktop/code/SnarfCode/homelab-gitops-repo/manifests/spaceport/semaphore/semaphore.yaml" if False else "k3s kubectl get pods -n infrastructure")
# Pull latest manifest via git in gitops repo or apply directly
with open(r"C:\Users\Snarf\OneDrive\Desktop\code\SnarfCode\homelab-gitops-repo\manifests\spaceport\semaphore\semaphore.yaml", "r", encoding="utf-8") as f:
manifest_content = f.read()
stdin, stdout, stderr = ssh.exec_command("k3s kubectl apply -f - --server-side=true")
stdin.write(manifest_content)
stdin.channel.shutdown_write()
print("Apply output:", stdout.read().decode())
err = stderr.read().decode()
if err: print("Apply stderr:", err)
# Restart pod to pick up new spec immediately
stdin, stdout, stderr = ssh.exec_command("k3s kubectl rollout restart deployment/semaphore -n infrastructure")
print(stdout.read().decode())
print("Waiting for rollout...")
stdin, stdout, stderr = ssh.exec_command("k3s kubectl rollout status deployment/semaphore -n infrastructure --timeout=90s")
print(stdout.read().decode())
# Verify new pod env
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n infrastructure -l app.kubernetes.io/name=semaphore -o jsonpath='{.items[0].metadata.name}'")
pod = stdout.read().decode().strip()
print(f"Active pod: {pod}")
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- env | grep SEMAPHORE_OIDC_PROVIDERS")
print("Live env:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,50 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect("140.44.4.71", username="root", key_filename=key_path, timeout=10)
py_script = """
import sqlite3, datetime
conn = sqlite3.connect('/var/lib/docker/volumes/nginx-proxy-manager_nginx-data/_data/database.sqlite')
cur = conn.cursor()
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
updates = [
(45, '140.44.4.34', 443, 'https'),
(41, '140.44.4.62', 443, 'https'),
(37, '140.44.4.34', 443, 'https'),
]
for pid, host, port, scheme in updates:
cur.execute('''UPDATE proxy_host
SET forward_host = ?, forward_port = ?, forward_scheme = ?,
allow_websocket_upgrade = 1, http2_support = 1, modified_on = ?
WHERE id = ?''', (host, port, scheme, now, pid))
print(f"Updated host {pid} -> {scheme}://{host}:{port}")
conn.commit()
for row in cur.execute('SELECT id, domain_names, forward_host, forward_port, forward_scheme FROM proxy_host WHERE id IN (45, 41, 37)'):
print("DB Row:", row)
conn.close()
"""
# Write script to /tmp/update_db.py on NPM host and execute
sftp = c.open_sftp()
with sftp.open('/tmp/update_db.py', 'w') as f:
f.write(py_script)
sftp.close()
stdin, stdout, stderr = c.exec_command("python3 /tmp/update_db.py")
print(stdout.read().decode())
err = stderr.read().decode()
if err: print("Err:", err)
c.close()

View File

@@ -0,0 +1,29 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def run_ssh(host, user, cmd):
print(f"=== SSH {user}@{host}: {cmd} ===")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(host, username=user, key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
if out:
print(out)
if err:
print("STDERR:", err)
except Exception as e:
print(f"Error connecting to {host}: {e}")
finally:
ssh.close()
# Test Aerodrome with root
run_ssh('140.44.4.11', 'root', 'k3s kubectl get pods -n dns -o wide')
run_ssh('140.44.4.11', 'root', 'k3s kubectl get svc -n dns -o wide')

View File

@@ -0,0 +1,14 @@
import urllib.request
import ssl
import json
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request("https://semaphore.snarfnet.net/api/auth/login", headers={"User-Agent": "Mozilla/5.0"})
resp = urllib.request.urlopen(req, context=ctx)
print("Status:", resp.status)
data = json.loads(resp.read().decode('utf-8'))
print("Data:", json.dumps(data, indent=2))

View File

@@ -0,0 +1,19 @@
import urllib.request
import ssl
import json
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
url = "https://authentik.snarfnet.net/application/o/semaphore/.well-known/openid-configuration"
try:
resp = urllib.request.urlopen(urllib.request.Request(url), context=ctx)
data = json.loads(resp.read().decode())
print("Issuer:", data.get('issuer'))
print("Auth endpoint:", data.get('authorization_endpoint'))
print("Token endpoint:", data.get('token_endpoint'))
print("Userinfo endpoint:", data.get('userinfo_endpoint'))
except Exception as e:
print("Error:", e)

View File

@@ -0,0 +1,20 @@
import urllib.request
import ssl
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
for host in ["https://sn-nas.snarfnet.net", "https://snarfnas.snarfnet.net"]:
try:
req = urllib.request.Request(host, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, context=ctx, timeout=8) as resp:
print(f"{host} -> Status: {resp.status}, Content-Length: {resp.headers.get('Content-Length')}")
body = resp.read().decode('utf-8', errors='replace')
print("Title/Body preview:", body[:300].replace('\n', ' '))
except Exception as e:
print(f"{host} -> Error: {e}")

View File

@@ -0,0 +1,24 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n infrastructure -l app.kubernetes.io/name=semaphore -o jsonpath='{.items[0].metadata.name}'")
pod = stdout.read().decode().strip()
print(f"Active pod: {pod}")
def test_url(url):
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- curl -s -I '{url}'")
print(f"Curl {url}:\n{stdout.read().decode()}")
test_url("http://localhost:3000/api/auth/oidc/authentik/login")
test_url("http://localhost:3000/api/auth/oidc/authentik/redirect")
test_url("http://localhost:3000/api/auth/oidc/authentik/redirect/")
ssh.close()

View File

@@ -0,0 +1,36 @@
import urllib.request
import ssl
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
opener = urllib.request.build_opener(NoRedirectHandler)
tests = [
"https://longhorn.aerodrome.snarfnet.net",
"https://semaphore.snarfnet.net",
"https://uptime.snarfnet.net",
"https://pihole1.snarfnet.net",
"https://pihole2.snarfnet.net"
]
print("=== Testing External/NPM Endpoints ===")
for url in tests:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
resp = opener.open(req, timeout=8)
print(f"✓ {url} -> {resp.status} {resp.reason}")
except urllib.error.HTTPError as e:
loc = e.headers.get('Location', '')
print(f"✓ {url} -> {e.code} {e.reason} (Location: {loc[:80]}...)")
except Exception as e:
print(f"✗ {url} -> Error: {e}")

View File

@@ -0,0 +1,26 @@
import urllib.request
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
opener = urllib.request.build_opener(NoRedirectHandler)
url = "https://semaphore.snarfnet.net/api/auth/oidc/authentik/redirect?code=sample&state=sample"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
resp = opener.open(req)
print("Status:", resp.status)
print("Headers:", resp.headers)
except urllib.error.HTTPError as e:
print("HTTPError:", e.code, e.reason)
print("Location:", e.headers.get('Location'))
print("Body:", e.read().decode('utf-8', errors='replace'))
except Exception as e:
print("Error:", e)

View File

@@ -0,0 +1,26 @@
import urllib.request
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
opener = urllib.request.build_opener(NoRedirectHandler)
url = "https://semaphore.snarfnet.net/api/auth/oidc/authentik/login"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
resp = opener.open(req)
print("Status:", resp.status)
print("Headers:", resp.headers)
except urllib.error.HTTPError as e:
print("HTTPError:", e.code, e.reason)
print("Location:", e.headers.get('Location'))
print("Body:", e.read().decode('utf-8', errors='replace'))
except Exception as e:
print("Error:", e)

View File

@@ -1,31 +1,24 @@
import urllib.request
import urllib.parse
import http.cookiejar
import paramiko
import sys
def test_login(ip):
cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
# 1. Fetch login page to get CSRF/session
login_url = f"http://{ip}/admin/login"
req = urllib.request.Request(login_url, headers={"User-Agent": "Mozilla/5.0"})
with opener.open(req) as resp:
html = resp.read().decode('utf-8')
# 2. Submit password
post_data = urllib.parse.urlencode({"pw": "snarfnetadmin", "login": "Log in (uses cookie)"}).encode('ascii')
req_post = urllib.request.Request(f"http://{ip}/admin/login", data=post_data, headers={"User-Agent": "Mozilla/5.0", "Referer": login_url})
try:
with opener.open(req_post) as resp:
content = resp.read().decode('utf-8')
if "Wrong password" in content:
print(f"Server {ip}: WRONG PASSWORD!")
else:
print(f"Server {ip}: LOGIN SUCCESSFUL! Redirected to {resp.geturl()}")
except Exception as e:
print(f"Server {ip}: Exception during login: {e}")
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
test_login("140.44.4.36")
test_login("140.44.4.65")
def test_login(ip, name):
print(f"\n--- Testing Pi-hole auth on {name} ({ip}) ---")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ip, username='root', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("k3s kubectl get pods -n dns -l app=pihole -o jsonpath='{.items[0].metadata.name}'")
pod = stdout.read().decode().strip()
# Test login via curl to Pi-hole API
cmd = f'k3s kubectl exec -n dns {pod} -- curl -s -X POST -H "Content-Type: application/json" -d \'{{"password":"snarfnetadmin"}}\' http://localhost/api/auth'
stdin, stdout, stderr = ssh.exec_command(cmd)
print("API Auth response:", stdout.read().decode())
ssh.close()
test_login('140.44.4.11', 'Aerodrome')
test_login('140.44.4.41', 'Spaceport')

View File

@@ -0,0 +1,19 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("140.44.4.41", username="root", key_filename=key_path, timeout=10)
pod = "semaphore-98778cd46-sjs7h"
stdin, stdout, stderr = ssh.exec_command(f"k3s kubectl exec -n infrastructure {pod} -- curl -s -I https://authentik.snarfnet.net/application/o/semaphore/.well-known/openid-configuration")
print("Response from inside Semaphore pod:\n", stdout.read().decode())
err = stderr.read().decode()
if err: print("Error:\n", err)
ssh.close()

View File

@@ -0,0 +1,27 @@
import urllib.request
import ssl
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
opener = urllib.request.build_opener(NoRedirectHandler)
url = "https://semaphore.snarfnet.net/api/auth/oidc/authentik/redirect/?code=samplecode&state=samplestate"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
resp = opener.open(req, timeout=8)
print(f"Status: {resp.status} {resp.reason}")
except urllib.error.HTTPError as e:
print(f"Status: {e.code} {e.reason}")
print(f"Location: {e.headers.get('Location')}")
body = e.read().decode('utf-8', errors='replace')
print("Body:", body[:200])

View File

@@ -0,0 +1,32 @@
import urllib.request
import ssl
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
opener = urllib.request.build_opener(NoRedirectHandler)
tests = [
"https://semaphore.snarfnet.net/",
"https://semaphore.snarfnet.net/auth/oidc/authentik/auth/login",
"https://semaphore.snarfnet.net/auth/oidc/authentik/redirect?code=sample&state=sample",
]
for url in tests:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
resp = opener.open(req, timeout=8)
print(f"✓ {url} -> {resp.status} {resp.reason}")
except urllib.error.HTTPError as e:
print(f"✓ {url} -> {e.code} {e.reason} (Location: {e.headers.get('Location')})")
except Exception as e:
print(f"✗ {url} -> Error: {e}")

View File

@@ -0,0 +1,17 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('140.44.4.71', username='snarf', key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command("curl -s -I 'http://140.44.4.61:100/api/auth/oidc/authentik/redirect/?code=test&state=test' -H 'Host: semaphore.snarfnet.net'")
print("Curl to 140.44.4.61:100:\n", stdout.read().decode())
stdin, stdout, stderr = ssh.exec_command("curl -k -s -I 'https://140.44.4.70/api/auth/oidc/authentik/redirect/?code=test&state=test' -H 'Host: semaphore.snarfnet.net'")
print("Curl to NPM VIP 140.44.4.70:\n", stdout.read().decode())
ssh.close()

View File

@@ -0,0 +1,29 @@
import paramiko
import sys
sys.stdout.reconfigure(encoding='utf-8')
key_path = r'C:\Users\Snarf\.ssh\id_ed25519_snarfnet'
def run_ssh(host, user, cmd):
print(f"=== SSH {user}@{host}: {cmd} ===")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(host, username=user, key_filename=key_path, timeout=10)
stdin, stdout, stderr = ssh.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
if out:
print(out)
if err:
print("STDERR:", err)
except Exception as e:
print(f"Error connecting to {host}: {e}")
finally:
ssh.close()
# Test Spaceport with root
run_ssh('140.44.4.41', 'root', 'k3s kubectl get pods -n dns -o wide')
run_ssh('140.44.4.41', 'root', 'k3s kubectl get svc -n dns -o wide')

33
scripts/test_urls.py Normal file
View File

@@ -0,0 +1,33 @@
import urllib.request
import ssl
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
urls = [
"https://pihole1.snarfnet.net",
"https://pihole2.snarfnet.net",
"https://dns1.snarfnet.net",
"https://dns2.snarfnet.net"
]
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
opener = urllib.request.build_opener(NoRedirectHandler)
for url in urls:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
resp = opener.open(req, timeout=5)
print(f"{url} -> {resp.status} {resp.reason}")
except urllib.error.HTTPError as e:
print(f"{url} -> {e.code} {e.reason} (Location: {e.headers.get('Location')})")
except Exception as e:
print(f"{url} -> Error: {e}")

View File

@@ -0,0 +1,38 @@
import urllib.request
import ssl
import json
import sys
sys.stdout.reconfigure(encoding='utf-8')
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
token = "SnarfNetPermanentAutomationToken2026"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
provider_update = {
"redirect_uris": [
{"matching_mode": "strict", "url": "https://semaphore.snarfnet.net/api/auth/oidc/authentik/redirect"},
{"matching_mode": "strict", "url": "https://semaphore.snarfnet.net/api/auth/oidc/authentik/redirect/"},
{"matching_mode": "regex", "url": "https://semaphore\\.snarfnet\\.net/.*"}
]
}
req = urllib.request.Request(
"https://authentik.snarfnet.net/api/v3/providers/oauth2/10/",
headers=headers,
data=json.dumps(provider_update).encode('utf-8'),
method="PATCH"
)
try:
with urllib.request.urlopen(req, context=ctx, timeout=8) as resp:
res = json.loads(resp.read().decode())
print("✓ Authentik Provider 10 updated successfully:", res.get('redirect_uris'))
except Exception as e:
print("✗ Failed to update Authentik Provider 10:", e)