namecheap-api/api_server.py

#!/usr/bin/env python3
"""
Namecheap REST API Server for OpenAI Function Calling
"""

import os
from typing import Optional
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Header, Depends
from pydantic import BaseModel

from namecheap import NamecheapAPI, NamecheapConfig, RegistrantInfo

load_dotenv()

# API Key for authentication
API_KEY = os.getenv("API_SERVER_KEY", "")


def verify_api_key(x_api_key: str = Header(None)):
    if API_KEY and x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")


app = FastAPI(
    title="Namecheap API",
    description="REST API for Namecheap domain management",
    version="1.0.0",
    dependencies=[Depends(verify_api_key)],
)

# Initialize Namecheap API
config = NamecheapConfig(
    api_user=os.getenv("NAMECHEAP_API_USER", ""),
    api_key=os.getenv("NAMECHEAP_API_KEY", ""),
    username=os.getenv("NAMECHEAP_USERNAME", ""),
    client_ip=os.getenv("NAMECHEAP_CLIENT_IP", ""),
    sandbox=os.getenv("NAMECHEAP_SANDBOX", "false").lower() == "true",
)
api = NamecheapAPI(config)


# ===== Models =====

class DomainCheckRequest(BaseModel):
    domains: list[str]

class DomainRenewRequest(BaseModel):
    domain: str
    years: int = 1

class NameserverRequest(BaseModel):
    domain: str
    nameservers: list[str]


class GlueRecordRequest(BaseModel):
    nameserver: str
    ip: str


class GlueRecordUpdateRequest(BaseModel):
    nameserver: str
    old_ip: str
    ip: str

class ContactRequest(BaseModel):
    domain: str
    organization: str = ""
    first_name: str
    last_name: str
    address1: str
    address2: str = ""
    city: str
    state_province: str
    postal_code: str
    country: str
    phone: str
    email: str


# ===== Domain Endpoints =====

@app.get("/domains")
def list_domains(page: int = 1, page_size: int = 20):
    """Get list of domains in account"""
    return api.domains_get_list(page=page, page_size=page_size)


@app.post("/domains/check")
def check_domains(req: DomainCheckRequest):
    """Check domain availability"""
    return api.domains_check(req.domains)


@app.get("/domains/{domain}")
def get_domain_info(domain: str):
    """Get detailed info about a domain"""
    return api.domains_get_info(domain)


@app.post("/domains/{domain}/renew")
def renew_domain(domain: str, years: int = 1):
    """Renew a domain. WARNING: This will charge your account."""
    return api.domains_renew(domain, years=years)


@app.get("/domains/{domain}/contacts")
def get_domain_contacts(domain: str):
    """Get domain contact information"""
    return api.domains_get_contacts(domain)


@app.put("/domains/{domain}/contacts")
def set_domain_contacts(domain: str, req: ContactRequest):
    """Set domain contact information"""
    registrant = RegistrantInfo(
        organization=req.organization,
        first_name=req.first_name,
        last_name=req.last_name,
        address1=req.address1,
        address2=req.address2,
        city=req.city,
        state_province=req.state_province,
        postal_code=req.postal_code,
        country=req.country,
        phone=req.phone,
        email=req.email,
    )
    success = api.domains_set_contacts(domain, registrant)
    return {"success": success}


# ===== DNS Endpoints =====

@app.get("/dns/{domain}/nameservers")
def get_nameservers(domain: str):
    """Get nameserver information for a domain"""
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    return api.dns_get_list(sld, tld)


@app.put("/dns/{domain}/nameservers")
def set_nameservers(domain: str, req: NameserverRequest):
    """Set custom nameservers for a domain"""
    from namecheap import NamecheapError
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    try:
        success = api.dns_set_custom(sld, tld, req.nameservers)
        return {"success": success, "nameservers": req.nameservers}
    except NamecheapError as e:
        error_msg = str(e)
        if "subordinate" in error_msg.lower() or "non existen" in error_msg.lower():
            raise HTTPException(
                status_code=400,
                detail=f"네임서버 {req.nameservers}는 등록되지 않았습니다. 자기 도메인을 네임서버로 사용하려면 먼저 Child Nameserver(글루 레코드)를 IP 주소와 함께 등록해야 합니다."
            )
        raise HTTPException(status_code=400, detail=error_msg)


@app.post("/dns/{domain}/nameservers/default")
def set_default_nameservers(domain: str):
    """Set default Namecheap nameservers"""
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    success = api.dns_set_default(sld, tld)
    return {"success": success}


@app.get("/dns/{domain}/records")
def get_dns_records(domain: str):
    """Get DNS records. Only works if using Namecheap DNS."""
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    return api.dns_get_hosts(sld, tld)


# ===== Glue Record (Child Nameserver) Endpoints =====

@app.post("/dns/{domain}/glue")
def create_glue_record(domain: str, req: GlueRecordRequest):
    """Create a glue record (child nameserver) for a domain"""
    from namecheap import NamecheapError
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    try:
        return api.ns_create(sld, tld, req.nameserver, req.ip)
    except NamecheapError as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.get("/dns/{domain}/glue/{nameserver}")
def get_glue_record(domain: str, nameserver: str):
    """Get info about a glue record (child nameserver)"""
    from namecheap import NamecheapError
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    try:
        return api.ns_get_info(sld, tld, nameserver)
    except NamecheapError as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.put("/dns/{domain}/glue")
def update_glue_record(domain: str, req: GlueRecordUpdateRequest):
    """Update a glue record (child nameserver) IP address"""
    from namecheap import NamecheapError
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    try:
        return api.ns_update(sld, tld, req.nameserver, req.old_ip, req.ip)
    except NamecheapError as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.delete("/dns/{domain}/glue/{nameserver}")
def delete_glue_record(domain: str, nameserver: str):
    """Delete a glue record (child nameserver)"""
    from namecheap import NamecheapError
    parts = domain.rsplit(".", 1)
    if len(parts) != 2:
        raise HTTPException(status_code=400, detail="Invalid domain format")
    sld, tld = parts
    try:
        return api.ns_delete(sld, tld, nameserver)
    except NamecheapError as e:
        raise HTTPException(status_code=400, detail=str(e))


# ===== Account Endpoints =====

@app.get("/account/balance")
def get_balance():
    """Get account balance"""
    return api.users_get_balances()


@app.get("/prices/{tld}")
def get_price(tld: str):
    """Get registration price for a TLD"""
    from db import get_prices, init_db
    init_db()

    prices = get_prices()
    for p in prices:
        if p["tld"] == tld:
            return {"tld": tld, "usd": p["usd"], "krw": p["krw"]}

    raise HTTPException(status_code=404, detail=f"TLD '{tld}' not found")


@app.get("/prices")
def list_prices(limit: int = 50):
    """Get all TLD prices"""
    from db import get_prices, init_db
    init_db()
    return get_prices()[:limit]


# ===== OpenAI Schema =====

@app.get("/openai/schema")
def get_openai_schema():
    """Get OpenAI Function Calling schema"""
    return {
        "functions": [
            {
                "name": "list_domains",
                "description": "Get list of domains in account",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "page": {"type": "integer", "default": 1},
                        "page_size": {"type": "integer", "default": 20}
                    }
                }
            },
            {
                "name": "check_domains",
                "description": "Check domain availability",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "domains": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "List of domains to check"
                        }
                    },
                    "required": ["domains"]
                }
            },
            {
                "name": "get_domain_info",
                "description": "Get detailed info about a domain",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "domain": {"type": "string"}
                    },
                    "required": ["domain"]
                }
            },
            {
                "name": "get_nameservers",
                "description": "Get nameserver information for a domain",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "domain": {"type": "string"}
                    },
                    "required": ["domain"]
                }
            },
            {
                "name": "set_nameservers",
                "description": "Set custom nameservers for a domain",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "domain": {"type": "string"},
                        "nameservers": {
                            "type": "array",
                            "items": {"type": "string"}
                        }
                    },
                    "required": ["domain", "nameservers"]
                }
            },
            {
                "name": "get_balance",
                "description": "Get account balance",
                "parameters": {"type": "object", "properties": {}}
            },
            {
                "name": "get_price",
                "description": "Get registration price for a TLD",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "tld": {"type": "string", "description": "TLD like com, net, io"}
                    },
                    "required": ["tld"]
                },
            },
            {
                "name": "register_domain",
                "description": "Register a new domain. WARNING: This will charge your account.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "domain": {"type": "string", "description": "Domain to register (e.g., example.com)"},
                        "years": {"type": "integer", "default": 1, "description": "Registration period (1-10 years)"}
                    },
                    "required": ["domain"]
                }
            }
        ]
    }




# ===== Domain Registration =====

class DomainRegisterRequest(BaseModel):
    domain: str
    years: int = 1
    add_whois_guard: bool = True
    # Registrant info (optional - uses default if not provided)
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    organization: Optional[str] = None
    address1: Optional[str] = None
    address2: Optional[str] = None
    city: Optional[str] = None
    state_province: Optional[str] = None
    postal_code: Optional[str] = None
    country: Optional[str] = None
    phone: Optional[str] = None
    email: Optional[str] = None


# Default registrant info from environment
def get_default_registrant() -> RegistrantInfo:
    return RegistrantInfo(
        first_name=os.getenv("REGISTRANT_FIRST_NAME", ""),
        last_name=os.getenv("REGISTRANT_LAST_NAME", ""),
        organization=os.getenv("REGISTRANT_ORGANIZATION", ""),
        address1=os.getenv("REGISTRANT_ADDRESS1", ""),
        address2=os.getenv("REGISTRANT_ADDRESS2", ""),
        city=os.getenv("REGISTRANT_CITY", ""),
        state_province=os.getenv("REGISTRANT_STATE_PROVINCE", ""),
        postal_code=os.getenv("REGISTRANT_POSTAL_CODE", ""),
        country=os.getenv("REGISTRANT_COUNTRY", "JP"),
        phone=os.getenv("REGISTRANT_PHONE", ""),
        email=os.getenv("REGISTRANT_EMAIL", ""),
    )


@app.post("/domains/register")
def register_domain(req: DomainRegisterRequest):
    """
    Register a new domain. WARNING: This will charge your account.
    
    If registrant info is not provided, uses default from environment.
    """
    from namecheap import NamecheapError
    
    # Use provided info or fall back to defaults
    default = get_default_registrant()
    registrant = RegistrantInfo(
        first_name=req.first_name or default.first_name,
        last_name=req.last_name or default.last_name,
        organization=req.organization or default.organization,
        address1=req.address1 or default.address1,
        address2=req.address2 or default.address2,
        city=req.city or default.city,
        state_province=req.state_province or default.state_province,
        postal_code=req.postal_code or default.postal_code,
        country=req.country or default.country,
        phone=req.phone or default.phone,
        email=req.email or default.email,
    )
    
    # Validate required fields
    required = ["first_name", "last_name", "address1", "city", "state_province", 
                "postal_code", "country", "phone", "email"]
    missing = [f for f in required if not getattr(registrant, f)]
    if missing:
        raise HTTPException(
            status_code=400, 
            detail=f"Missing required registrant fields: {', '.join(missing)}. Set DEFAULT_* env vars or provide in request."
        )
    
    try:
        result = api.domains_create(
            domain=req.domain,
            registrant=registrant,
            years=req.years,
            add_whois_guard=req.add_whois_guard,
        )
        return result
    except NamecheapError as e:
        raise HTTPException(status_code=400, detail=str(e))


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)