BurgerGames/NetBirdMSP-Appliance
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add SSL certificate mode: Let's Encrypt or Wildcard per NPM

Browse Source 
Settings > NPM Integration now allows choosing between per-customer
Let's Encrypt certificates (default) or a shared wildcard certificate
already uploaded in NPM. Includes backend, frontend UI, and i18n support.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Sascha Lustenberger
2026-02-09 00:01:28 +01:00
parent 3d28f13054
commit c7fc4758e3
12 changed files with 274 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/database.py
View File

@@ -83,6 +83,8 @@ def _run_migrations() -> None:
("system_config", "mfa_enabled", "BOOLEAN DEFAULT 0"),
("users", "totp_secret_encrypted", "TEXT"),
("users", "totp_enabled", "BOOLEAN DEFAULT 0"),
("system_config", "ssl_mode", "TEXT DEFAULT 'letsencrypt'"),
("system_config", "wildcard_cert_id", "INTEGER"),
]
for table, column, col_type in migrations:
if not _has_column(table, column):

4
app/models.py
View File

@@ -161,6 +161,8 @@ class SystemConfig(Base):
)
branding_logo_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
default_language: Mapped[Optional[str]] = mapped_column(String(10), default="en")
ssl_mode: Mapped[str] = mapped_column(String(20), default="letsencrypt")
wildcard_cert_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
mfa_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
azure_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
azure_tenant_id: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
@@ -194,6 +196,8 @@ class SystemConfig(Base):
"branding_subtitle": self.branding_subtitle or "Multi-Tenant Management Platform",
"branding_logo_path": self.branding_logo_path,
"default_language": self.default_language or "en",
"ssl_mode": self.ssl_mode or "letsencrypt",
"wildcard_cert_id": self.wildcard_cert_id,
"mfa_enabled": bool(self.mfa_enabled),
"azure_enabled": bool(self.azure_enabled),
"azure_tenant_id": self.azure_tenant_id or "",

35
app/routers/settings.py
View File

@@ -129,6 +129,41 @@ async def test_npm(
return result
@router.get("/npm-certificates")
async def list_npm_certificates(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""List all SSL certificates configured in NPM.
Used by the frontend to populate the wildcard certificate dropdown.
Returns:
List of certificate dicts with id, domain_names, provider, expires_on, is_wildcard.
"""
config = get_system_config(db)
if not config:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="System configuration not initialized.",
)
if not config.npm_api_url or not config.npm_api_email or not config.npm_api_password:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="NPM API URL or credentials not configured.",
)
result = await npm_service.list_certificates(
config.npm_api_url, config.npm_api_email, config.npm_api_password
)
if "error" in result:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=result["error"],
)
return result["certificates"]
@router.get("/branding")
async def get_branding(db: Session = Depends(get_db)):
"""Public endpoint — returns branding info for the login page (no auth required)."""

2
app/services/netbird_service.py
View File

@@ -277,6 +277,8 @@ async def deploy_customer(db: Session, customer_id: int) -> dict[str, Any]:
forward_host=forward_host,
forward_port=dashboard_port,
admin_email=config.admin_email,
ssl_mode=config.ssl_mode,
wildcard_cert_id=config.wildcard_cert_id,
)
npm_proxy_id = npm_result.get("proxy_id")
if npm_result.get("error"):

96
app/services/npm_service.py
View File

@@ -112,6 +112,45 @@ async def test_npm_connection(api_url: str, email: str, password: str) -> dict[s
return {"ok": False, "message": f"Unexpected error: {exc}"}
async def list_certificates(api_url: str, email: str, password: str) -> dict[str, Any]:
"""Fetch all SSL certificates from NPM.
Args:
api_url: NPM API base URL.
email: NPM login email.
password: NPM login password.
Returns:
Dict with ``certificates`` list on success, or ``error`` on failure.
"""
try:
async with httpx.AsyncClient(timeout=NPM_TIMEOUT) as client:
token = await _npm_login(client, api_url, email, password)
headers = {"Authorization": f"Bearer {token}"}
resp = await client.get(f"{api_url}/nginx/certificates", headers=headers)
if resp.status_code == 200:
result = []
for cert in resp.json():
domains = cert.get("domain_names", [])
result.append({
"id": cert.get("id"),
"domain_names": domains,
"provider": cert.get("provider", "unknown"),
"expires_on": cert.get("expires_on"),
"is_wildcard": any(d.startswith("*.") for d in domains),
})
return {"certificates": result}
return {"error": f"NPM returned {resp.status_code}: {resp.text[:200]}"}
except RuntimeError as exc:
return {"error": str(exc)}
except httpx.ConnectError:
return {"error": "Connection refused. Is NPM running and reachable?"}
except httpx.TimeoutException:
return {"error": "Connection timed out."}
except Exception as exc:
return {"error": f"Unexpected error: {exc}"}
async def _find_cert_by_domain(
client: httpx.AsyncClient, api_url: str, headers: dict, domain: str
) -> int | None:
@@ -169,6 +208,8 @@ async def create_proxy_host(
forward_host: str,
forward_port: int = 80,
admin_email: str = "",
ssl_mode: str = "letsencrypt",
wildcard_cert_id: int | None = None,
) -> dict[str, Any]:
"""Create a proxy host entry in NPM with SSL for a customer.
@@ -265,7 +306,10 @@ async def create_proxy_host(
return {"error": error_msg}
# Step 2: Request SSL certificate and enable HTTPS
ssl_ok = await _request_ssl(client, api_url, headers, proxy_id, domain, admin_email)
ssl_ok = await _request_ssl(
client, api_url, headers, proxy_id, domain, admin_email,
ssl_mode=ssl_mode, wildcard_cert_id=wildcard_cert_id,
)
return {"proxy_id": proxy_id, "ssl": ssl_ok}
except RuntimeError as exc:
@@ -283,13 +327,14 @@ async def _request_ssl(
proxy_id: int,
domain: str,
admin_email: str,
ssl_mode: str = "letsencrypt",
wildcard_cert_id: int | None = None,
) -> bool:
"""Request a Let's Encrypt SSL certificate and enable HTTPS on the proxy host.
"""Request an SSL certificate and enable HTTPS on the proxy host.
Flow:
1. Create LE certificate via NPM API (HTTP-01 validation, up to 120s)
2. Assign certificate to the proxy host
3. Enable ssl_forced + hsts on the proxy host
Supports two modes:
- ``letsencrypt``: Create a per-domain LE certificate (HTTP-01 validation).
- ``wildcard``: Assign a pre-existing wildcard certificate from NPM.
Args:
client: httpx client (already authenticated).
@@ -298,10 +343,49 @@ async def _request_ssl(
proxy_id: The proxy host ID.
domain: The domain to certify.
admin_email: Contact email for LE.
ssl_mode: ``"letsencrypt"`` or ``"wildcard"``.
wildcard_cert_id: NPM certificate ID for wildcard mode.
Returns:
True if SSL was successfully enabled, False otherwise.
"""
# Wildcard mode: assign the pre-existing wildcard cert directly
if ssl_mode == "wildcard" and wildcard_cert_id:
logger.info(
"Wildcard mode: assigning cert id=%s to proxy host %s for %s",
wildcard_cert_id, proxy_id, domain,
)
ssl_update = {
"certificate_id": wildcard_cert_id,
"ssl_forced": True,
"hsts_enabled": True,
"http2_support": True,
}
try:
update_resp = await client.put(
f"{api_url}/nginx/proxy-hosts/{proxy_id}",
json=ssl_update,
headers=headers,
)
if update_resp.status_code in (200, 201):
logger.info(
"SSL enabled on proxy host %s (wildcard cert_id=%s)",
proxy_id, wildcard_cert_id,
)
return True
logger.error(
"Failed to assign wildcard cert %s to proxy host %s: HTTP %s — %s",
wildcard_cert_id, proxy_id,
update_resp.status_code, update_resp.text[:300],
)
return False
except Exception as exc:
logger.error(
"Failed to assign wildcard cert to proxy host %s: %s", proxy_id, exc,
)
return False
# Let's Encrypt mode (default)
if not admin_email:
logger.warning("No admin email set — skipping SSL certificate for %s", domain)
return False

4
app/utils/config.py
View File

@@ -31,6 +31,8 @@ class AppConfig:
docker_network: str
relay_base_port: int
dashboard_base_port: int
ssl_mode: str
wildcard_cert_id: int | None
# Environment-level settings (not stored in DB)
@@ -79,4 +81,6 @@ def get_system_config(db: Session) -> Optional[AppConfig]:
docker_network=row.docker_network,
relay_base_port=row.relay_base_port,
dashboard_base_port=getattr(row, "dashboard_base_port", 9000) or 9000,
ssl_mode=getattr(row, "ssl_mode", "letsencrypt") or "letsencrypt",
wildcard_cert_id=getattr(row, "wildcard_cert_id", None),
)

13
app/utils/validators.py
View File

@@ -126,12 +126,25 @@ class SystemConfigUpdate(BaseModel):
branding_name: Optional[str] = Field(None, max_length=255)
branding_subtitle: Optional[str] = Field(None, max_length=255)
default_language: Optional[str] = Field(None, max_length=10)
ssl_mode: Optional[str] = Field(None, max_length=20)
wildcard_cert_id: Optional[int] = Field(None, ge=0)
mfa_enabled: Optional[bool] = None
azure_enabled: Optional[bool] = None
azure_tenant_id: Optional[str] = Field(None, max_length=255)
azure_client_id: Optional[str] = Field(None, max_length=255)
azure_client_secret: Optional[str] = None # encrypted before storage
@field_validator("ssl_mode")
@classmethod
def validate_ssl_mode(cls, v: Optional[str]) -> Optional[str]:
"""SSL mode must be 'letsencrypt' or 'wildcard'."""
if v is None:
return v
allowed = {"letsencrypt", "wildcard"}
if v not in allowed:
raise ValueError(f"ssl_mode must be one of: {', '.join(sorted(allowed))}")
return v
@field_validator("base_domain")
@classmethod
def validate_domain(cls, v: Optional[str]) -> Optional[str]: