test_api.py

#!/usr/bin/env python3
"""
Test script for Technitium DNS Server API connectivity.
This script tests the API connection without needing a real domain or Let's Encrypt.
"""

import sys
import os
import requests
import json
from pathlib import Path

def test_api_connection(api_url, api_token):
    """Test basic API connectivity."""
    print(f"Testing API connection to: {api_url}")
    print(f"Using API token: {api_token[:10]}..." if len(api_token) > 10 else f"Using API token: {api_token}")
    print("-" * 60)
    
    # Test 1: List zones
    print("\n[Test 1] Listing zones...")
    try:
        response = requests.get(
            f"{api_url}/api/zones/list",
            params={"token": api_token},
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()
        
        if data.get("status") == "ok":
            # Handle different response formats
            response_data = data.get("response", [])
            if isinstance(response_data, dict):
                # Response might be {"zones": [...]} or similar
                zones = response_data.get("zones", response_data.get("response", []))
                # Debug: show what keys are in the dict
                if not zones and response_data:
                    print(f"  Debug: Response dict keys: {list(response_data.keys())}")
            else:
                zones = response_data if isinstance(response_data, list) else []
            
            print(f"✓ Successfully connected to API")
            print(f"✓ Found {len(zones)} zone(s)")
            
            if zones:
                print("\nAvailable zones:")
                # Handle both string and dict formats
                zone_list = []
                for zone in zones:
                    if isinstance(zone, str):
                        # Zone is just a string (zone name)
                        zone_name = zone.rstrip(".")
                        print(f"  - {zone_name}")
                        zone_list.append({"name": zone_name})
                    elif isinstance(zone, dict):
                        # Zone is a dictionary with zone info
                        zone_name = zone.get("name", "unknown").rstrip(".")
                        zone_type = zone.get("type", "unknown")
                        print(f"  - {zone_name} (type: {zone_type})")
                        zone_list.append(zone)
                    else:
                        # Unknown format, try to convert
                        zone_name = str(zone).rstrip(".")
                        print(f"  - {zone_name}")
                        zone_list.append({"name": zone_name})
                zones = zone_list
            else:
                print("  (No zones found - this is okay for testing)")
            
            return True, zones
        else:
            error_msg = data.get("errorMessage") or data.get("response") or "Unknown error"
            print(f"✗ API returned error: {error_msg}")
            return False, []
            
    except requests.exceptions.ConnectionError:
        print(f"✗ Connection failed: Could not reach {api_url}")
        print("  Make sure Technitium DNS Server is running and accessible")
        return False, []
    except requests.exceptions.Timeout:
        print(f"✗ Connection timeout: Server did not respond in time")
        return False, []
    except requests.exceptions.RequestException as e:
        print(f"✗ Request failed: {e}")
        return False, []
    except json.JSONDecodeError:
        print(f"✗ Invalid JSON response from server")
        return False, []


def test_zone_operations(api_url, api_token, zones):
    """Test zone operations (add/delete TXT record)."""
    if not zones:
        print("\n[Test 2] Skipping zone operations (no zones available)")
        return True
    
    # Find a forward Primary zone (not reverse, not forwarder)
    test_zone = None
    for zone in zones:
        if isinstance(zone, dict):
            zone_name = zone.get("name", "").rstrip(".")
            zone_type = zone.get("type", "").lower()
        else:
            zone_name = str(zone).rstrip(".")
            zone_type = "primary"  # Assume primary if not specified
        
        # Skip reverse zones and forwarder zones
        if ".in-addr.arpa" in zone_name or ".ip6.arpa" in zone_name:
            continue
        if zone_type == "forwarder":
            continue
        if zone_name in ["localhost"]:
            continue
        
        # Use the first suitable forward Primary zone
        test_zone = zone_name
        break
    
    if not test_zone:
        print("\n[Test 2] Skipping zone operations (no suitable forward Primary zone found)")
        print("  (Reverse zones and Forwarder zones cannot have records added)")
        return True
    
    print(f"\n[Test 2] Testing zone operations on: {test_zone}")
    # Use full domain name for the record (subdomain.zone)
    test_record_name = f"_acme-challenge-test.{test_zone}"
    test_record_value = "test-validation-string-12345"
    
    # First, try to list records for this zone to verify the endpoint works
    print(f"  Testing zone access by listing records...")
    try:
        # According to API docs: /api/zones/records/get?token=x&domain=example.com&zone=example.com
        list_response = requests.get(
            f"{api_url}/api/zones/records/get",
            params={
                "token": api_token,
                "domain": test_zone,
                "zone": test_zone,
                "listZone": "true",
            },
            timeout=10,
        )
        if list_response.status_code == 200:
            list_data = list_response.json()
            if list_data.get("status") == "ok":
                print(f"  ✓ Zone access confirmed (can list records)")
            else:
                print(f"  ⚠ Zone list returned: {list_data.get('errorMessage', 'Unknown error')}")
        else:
            print(f"  ⚠ Zone list returned status: {list_response.status_code}")
            print(f"  This might indicate the zone name format is incorrect")
    except Exception as e:
        print(f"  ⚠ Could not list records: {e}")
    
    # Test adding a TXT record
    print(f"  Adding test TXT record: {test_record_name}")
    try:
        # Try without URL encoding first (dots might be safe in zone names)
        # If that doesn't work, we'll try with encoding
        from urllib.parse import quote
        
        # According to API docs: /api/zones/records/add?token=x&domain=example.com&zone=example.com
        # For TXT records, use 'text' parameter, not 'value'
        response = requests.post(
            f"{api_url}/api/zones/records/add",
            params={
                "token": api_token,
                "domain": test_record_name,
                "zone": test_zone,
                "type": "TXT",
                "ttl": 60,
                "text": test_record_value,
            },
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()
        
        if data.get("status") == "ok":
            print(f"  ✓ Successfully added TXT record")
            print(f"  Record added: {test_record_name} = {test_record_value}")
            print(f"  You can now check this record on your DNS server.")
            
            # Ask user if they want to delete the record now
            while True:
                response = input(f"  Delete the test record now? (y/n): ").strip().lower()
                if response in ['y', 'yes']:
                    break
                elif response in ['n', 'no']:
                    print(f"  Record left in place. You can delete it manually later.")
                    print(f"  Record: {test_record_name} = {test_record_value}")
                    return True
                else:
                    print(f"  Please enter 'y' or 'n'")
            
            # Test deleting the TXT record
            print(f"  Deleting test TXT record...")
            # According to API docs: /api/zones/records/delete?token=x&domain=example.com&zone=example.com&type=TXT&text=...
            # For TXT records, use 'text' parameter, not 'value'
            delete_response = requests.post(
                f"{api_url}/api/zones/records/delete",
                params={
                    "token": api_token,
                    "domain": test_record_name,
                    "zone": test_zone,
                    "type": "TXT",
                    "text": test_record_value,
                },
                timeout=10,
            )
            delete_response.raise_for_status()
            delete_data = delete_response.json()
            
            if delete_data.get("status") == "ok":
                print(f"  ✓ Successfully deleted TXT record")
                return True
            else:
                error_msg = delete_data.get("errorMessage") or delete_data.get("response") or "Unknown error"
                print(f"  ⚠ Could not delete record: {error_msg}")
                print(f"  (Record may need manual cleanup)")
                return True  # Still consider this a success
        else:
            error_msg = data.get("errorMessage") or data.get("response") or "Unknown error"
            print(f"  ✗ Failed to add TXT record: {error_msg}")
            return False
            
    except requests.exceptions.RequestException as e:
        print(f"  ✗ Request failed: {e}")
        return False


def load_credentials(credentials_file):
    """Load credentials from INI file."""
    api_url = None
    api_token = None
    
    try:
        with open(credentials_file, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    if '=' in line:
                        key, value = line.split('=', 1)
                        key = key.strip()
                        value = value.strip()
                        
                        if key == 'dns_technitium_api_url':
                            api_url = value
                        elif key == 'dns_technitium_api_token':
                            api_token = value
        
        if not api_url or not api_token:
            return None, None
        
        return api_url, api_token
    except FileNotFoundError:
        print(f"✗ Credentials file not found: {credentials_file}")
        return None, None
    except Exception as e:
        print(f"✗ Error reading credentials file: {e}")
        return None, None


def main():
    """Main test function."""
    print("=" * 60)
    print("Technitium DNS Server API Test Script")
    print("=" * 60)
    
    # Get credentials
    if len(sys.argv) > 1:
        credentials_file = sys.argv[1]
    else:
        # Try common locations
        possible_locations = [
            Path.home() / "technitium.ini",
            Path("/etc/letsencrypt/technitium.ini"),
            Path("technitium.ini"),
        ]
        
        credentials_file = None
        for loc in possible_locations:
            if loc.exists():
                credentials_file = str(loc)
                break
        
        if not credentials_file:
            print("\nUsage: python3 test_api.py [credentials_file]")
            print("\nOr set credentials via environment variables:")
            print("  export TECHNITIUM_API_URL=http://localhost:5380")
            print("  export TECHNITIUM_API_TOKEN=your-token")
            print("\nOr create a credentials file (technitium.ini):")
            print("  dns_technitium_api_url = http://localhost:5380")
            print("  dns_technitium_api_token = your-token-here")
            sys.exit(1)
    
    print(f"\nLoading credentials from: {credentials_file}")
    api_url, api_token = load_credentials(credentials_file)
    
    # Try environment variables as fallback
    if not api_url:
        api_url = os.environ.get("TECHNITIUM_API_URL", "http://localhost:5380")
    if not api_token:
        api_token = os.environ.get("TECHNITIUM_API_TOKEN")
    
    if not api_url or not api_token:
        print("✗ Missing API URL or token")
        sys.exit(1)
    
    # Run tests
    success, zones = test_api_connection(api_url, api_token)
    
    if not success:
        print("\n" + "=" * 60)
        print("✗ API connection test failed")
        print("=" * 60)
        sys.exit(1)
    
    # Test zone operations if we have zones
    if zones:
        test_zone_ops = test_zone_operations(api_url, api_token, zones)
    else:
        print("\n[Test 2] Skipping zone operations (no zones available)")
        print("  To test zone operations, create a zone in Technitium DNS Server first")
        test_zone_ops = True
    
    # Summary
    print("\n" + "=" * 60)
    if success and test_zone_ops:
        print("✓ All tests passed!")
        print("=" * 60)
        print("\nYour Technitium API is working correctly.")
        print("You can now use the certbot-dns-technitium plugin.")
        sys.exit(0)
    elif success:
        print("✓ API connection successful")
        print("⚠ Zone operations not tested (no zones available)")
        print("=" * 60)
        print("\nYour Technitium API is accessible.")
        print("Create a zone in Technitium DNS Server to test full functionality.")
        sys.exit(0)
    else:
        print("✗ Tests failed")
        print("=" * 60)
        sys.exit(1)


if __name__ == "__main__":
    main()