scripts/test_public_tunnel_credentials.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Check credential safety and registration over an isolated TLS server."""
import http.server
import json
import os
from pathlib import Path
import socket
import ssl
import subprocess
import tempfile
import threading
from types import SimpleNamespace
import unittest
from unittest.mock import patch
import public_tunnel_credentials as credentials
class Handler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
self.server.requests.append((self.path, dict(self.headers), body))
response = self.server.response
self.send_response(self.server.status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response)))
self.end_headers()
self.wfile.write(response)
def log_message(self, *_):
pass
class CredentialsTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.directory = tempfile.TemporaryDirectory()
cls.root = Path(os.path.realpath(cls.directory.name))
cls.cert = cls.root / "cert.pem"
key = cls.root / "key.pem"
subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
"-keyout", str(key), "-out", str(cls.cert), "-days", "1",
"-subj", "/CN=tunnel.firecrab.dev", "-addext", "subjectAltName=DNS:tunnel.firecrab.dev"],
check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(cls.cert, key)
cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
cls.server.socket = context.wrap_socket(cls.server.socket, server_side=True)
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
cls.thread.start()
@classmethod
def tearDownClass(cls):
cls.server.shutdown()
cls.server.server_close()
cls.thread.join()
cls.directory.cleanup()
def setUp(self):
self.first = self.root / "first.token"
self.second = self.root / "second.token"
self.first.unlink(missing_ok=True)
self.second.unlink(missing_ok=True)
self.first.write_text("a" * 64 + "\n")
self.second.write_text("b" * 64)
self.first.chmod(0o600)
self.second.chmod(0o400)
self.server.requests = []
self.server.status = 200
self.server.response = json.dumps({
"name": "qa-first", "url": "https://qa-first.firecrab.dev",
"heartbeat_url": "https://qa-first.firecrab.dev/__firecrab/heartbeat", "suspended": False,
}).encode()
original_connect = socket.create_connection
default_context = ssl.create_default_context
self.default_context = default_context
def connect(address, *args, **kwargs):
self.assertEqual(address, (credentials.HOST, 443))
return original_connect(("127.0.0.1", self.server.server_port), *args, **kwargs)
connection_patch = patch.object(socket, "create_connection", side_effect=connect)
context_patch = patch.object(credentials.ssl, "create_default_context", side_effect=lambda: default_context(cafile=str(self.cert)))
connection_patch.start()
context_patch.start()
self.addCleanup(connection_patch.stop)
self.addCleanup(context_patch.stop)
def pair(self):
return credentials.load_credentials([("qa-first", self.first), ("qa-second", self.second)])
def test_protected_pair_and_empty_authenticated_registration(self):
first, second = self.pair()
self.assertNotIn(first.token, repr(first))
self.assertNotIn(second.token, repr(second))
result = credentials.verify_registration(first)
self.assertEqual(result, {"label": "qa-first", "url": "https://qa-first.firecrab.dev", "status": 200})
path, headers, body = self.server.requests[0]
self.assertEqual(path, "/v1/register")
self.assertEqual(headers["X-Firecrab-Protocol"], "2")
self.assertEqual(headers["Authorization"], "Bearer " + first.token)
self.assertEqual(body, b"{}")
def test_rejects_shared_tokens_and_invalid_credentials(self):
self.second.chmod(0o600)
for entries in [
[("qa-first", self.first)],
[("qa-first", self.first), ("qa-first", self.second)],
[("qa-first", self.first), ("qa-second", self.first)],
[("console", self.first), ("qa-second", self.second)],
]:
with self.subTest(entries=entries), self.assertRaises(credentials.CredentialError):
credentials.load_credentials(entries)
self.second.write_text("a" * 64)
with self.assertRaises(credentials.CredentialError):
self.pair()
self.second.write_text("x" * 64)
with self.assertRaises(credentials.CredentialError):
self.pair()
self.second.write_text("b" * 64 + "\nextra")
with self.assertRaises(credentials.CredentialError):
self.pair()
def test_rejects_unsafe_file_permissions_and_links(self):
self.first.chmod(0o644)
with self.assertRaises(credentials.CredentialError):
self.pair()
self.first.chmod(0o600)
link = self.root / "token-link"
link.symlink_to(self.first)
with self.assertRaises(credentials.CredentialError):
credentials.load_credentials([("qa-first", link), ("qa-second", self.second)])
link.unlink()
link.symlink_to(self.root, target_is_directory=True)
with self.assertRaises(credentials.CredentialError):
credentials.load_credentials([("qa-first", link / self.first.name), ("qa-second", self.second)])
link.unlink()
os.link(self.first, link)
with self.assertRaises(credentials.CredentialError):
self.pair()
link.unlink()
fifo = self.root / "fifo"
os.mkfifo(fifo, 0o600)
with self.assertRaises(credentials.CredentialError):
credentials.load_credentials([("qa-first", fifo), ("qa-second", self.second)])
fifo.unlink()
def test_registration_checks_exact_identity_and_never_echoes_body(self):
first, _ = self.pair()
for response in [
{"name": "qa-second", "url": "https://qa-first.firecrab.dev", "heartbeat_url": "https://qa-first.firecrab.dev/__firecrab/heartbeat", "suspended": False},
{"name": "qa-first", "url": "https://qa-first.firecrab.dev", "heartbeat_url": "https://evil.invalid/", "suspended": False},
{"name": "qa-first", "url": "https://qa-first.firecrab.dev", "heartbeat_url": "https://qa-first.firecrab.dev/__firecrab/heartbeat", "suspended": True},
{"token": first.token},
]:
self.server.response = json.dumps(response).encode()
with self.subTest(response=list(response)), self.assertRaises(credentials.CredentialError) as failure:
credentials.verify_registration(first)
self.assertNotIn(first.token, str(failure.exception))
self.server.response = b"x" * (credentials.MAX_RESPONSE + 1)
with self.assertRaises(credentials.CredentialError):
credentials.verify_registration(first)
def test_rejects_token_file_owned_by_another_user(self):
original = os.fstat
def wrong_owner(fd):
info = original(fd)
return SimpleNamespace(st_mode=info.st_mode, st_nlink=info.st_nlink, st_uid=-1)
with patch.object(credentials.os, "fstat", side_effect=wrong_owner):
with self.assertRaises(credentials.CredentialError):
self.pair()
def test_revocation_requires_authentication_rejection(self):
first, _ = self.pair()
for status in (401, 410):
self.server.status = status
self.server.response = b'{}'
self.assertEqual(credentials.verify_revoked(first)["status"], status)
for status in (200, 403, 404, 429, 503):
self.server.status = status
with self.subTest(status=status), self.assertRaises(credentials.CredentialError):
credentials.verify_revoked(first)
def test_registration_refuses_untrusted_tls(self):
first, _ = self.pair()
with patch.object(credentials.ssl, "create_default_context", side_effect=self.default_context):
with self.assertRaises(credentials.CredentialError) as failure:
credentials.verify_registration(first)
self.assertNotIn(first.token, str(failure.exception))
self.assertEqual(self.server.requests, [])
if __name__ == "__main__":
unittest.main()