diff --git a/README.md b/README.md index d8f4bc2..efb3ccb 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# micro-license-server +# Micro License Server A very small, self-hosted license server intended for lower traffic projects. @@ -12,3 +12,18 @@ sed -i \ -e "s/API_CHANGEME/$(openssl rand -base64 48)/" \ .env ``` + +## API Endpoints + +All authenticated endpoints expect a `Bearer` token via the `Authorization` header that matches `API_KEY` from `.env`. Unless stated otherwise, responses are JSON. + +| Method | Path | Auth | Description | Key Request Parameters | Response Highlights | +| --- | --- | --- | --- | --- | --- | +| `GET` | `/` | No | Returns server metadata. | – | `{"version": "Micro License Server vX.Y.Z"}` | +| `POST` | `/license` | Yes | Generates a new license key. | Query: `is_active` (bool, default `true`), `expiration_date` (ISO 8601, optional) | `license_key`, `expiration_timestamp`, `is_active` | +| `GET` | `/is_valid` | No | Validates a license key and records last usage time. | Query: `license_key` (required) | Boolean validity | +| `POST` | `/license/{license_key}/disable` | Yes | Deactivates a license key. | Path: `license_key` | `license_key`, `is_active: false` | +| `POST` | `/license/{license_key}/enable` | Yes | Reactivates a license key. | Path: `license_key` | `license_key`, `is_active: true` | +| `POST` | `/license/{license_key}/expiration` | Yes | Sets or clears a license key expiration. | Path: `license_key`; Query: `expiration_date` (ISO 8601 or omit to clear) | `license_key`, `expiration_timestamp` (nullable) | +| `GET` | `/license/export` | Yes | Exports license inventory as CSV ordered by issue time. | – | CSV with `license_key,issue_timestamp,expiration_timestamp,is_active` | +| `GET` | `/history/export` | Yes | Exports audit history as CSV. | Query: `token` (optional substring filter) | CSV with `action,timestamp` | diff --git a/src/main.py b/src/main.py index 6bd042a..8bb366f 100644 --- a/src/main.py +++ b/src/main.py @@ -10,10 +10,17 @@ import psycopg import asyncio ENV = dotenv_values(".env") # .env file -VERSION = "v0.0.1" # version number +VERSION = "v1.0.0" # version number ALPHABET = string.ascii_uppercase + string.digits #alphabet and numbers for token generation -LICENSE_KEY_PARTS = 5 if "NUM_KEY_CHUNKS" not in ENV else ENV["NUM_KEY_CHUNKS"] #number of chunks in the new generated license keys -LICENSE_KEY_PART_LENGTH = 5 if "KEY_CHUNK_LENGTH" not in ENV else ENV["KEY_CHUNK_LENGTH"] #number of characters in each chunk +try: + LICENSE_KEY_PARTS = 5 if "NUM_KEY_CHUNKS" not in ENV else int(ENV["NUM_KEY_CHUNKS"]) #number of chunks in the new generated license keys +except: + LICENSE_KEY_PARTS = 5 + +try: + LICENSE_KEY_PART_LENGTH = 5 if "KEY_CHUNK_LENGTH" not in ENV else int(ENV["KEY_CHUNK_LENGTH"]) #number of characters in each chunk +except: + LICENSE_KEY_PART_LENGTH = 5 api_key = "" security = HTTPBearer(auto_error=False) diff --git a/tests/tests.py b/tests/tests.py index e69de29..7ac905b 100644 --- a/tests/tests.py +++ b/tests/tests.py @@ -0,0 +1,181 @@ +import csv +import os +from datetime import datetime, timedelta, timezone +from typing import Optional + +import pytest +import requests + + +def _resolve_base_url() -> str: + explicit = os.getenv("LICENSE_SERVER_BASE_URL") + if explicit: + return explicit.rstrip("/") + + host = os.getenv("LICENSE_SERVER_IP") + if host: + scheme = os.getenv("LICENSE_SERVER_SCHEME", "http") + port = os.getenv("LICENSE_SERVER_PORT", "8000") + return f"{scheme}://{host}:{port}".rstrip("/") + + return "http://127.0.0.1:8000" + + +def _resolve_timeout() -> float: + raw_timeout = os.getenv("LICENSE_SERVER_TIMEOUT", "5") + try: + return float(raw_timeout) + except ValueError: + return 5.0 + + +class LicenseServerClient: + """Minimal helper around requests for exercising the license server API.""" + + def __init__(self, base_url: str, api_key: str, timeout: float) -> None: + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.timeout = timeout + self.session = requests.Session() + + def close(self) -> None: + self.session.close() + + def request(self, method: str, path: str, *, auth: bool = True, **kwargs) -> requests.Response: + url = f"{self.base_url}{path}" + headers = kwargs.pop("headers", None) or {} + if auth and self.api_key: + headers = {"Authorization": f"Bearer {self.api_key}", **headers} + kwargs.setdefault("timeout", self.timeout) + return self.session.request(method=method, url=url, headers=headers, **kwargs) + + def server_info(self) -> requests.Response: + return self.request("get", "/", auth=False) + + def create_license( + self, + *, + is_active: bool = True, + expiration_iso: Optional[str] = None, + auth: bool = True, + ) -> requests.Response: + params = {} + if is_active is False: + params["is_active"] = "false" + if expiration_iso is not None: + params["expiration_date"] = expiration_iso + return self.request("post", "/license", params=params or None, auth=auth) + + def is_valid(self, license_key: str) -> requests.Response: + return self.request("get", "/is_valid", params={"license_key": license_key}, auth=False) + + def disable_license(self, license_key: str, *, auth: bool = True) -> requests.Response: + return self.request("post", f"/license/{license_key}/disable", auth=auth) + + def enable_license(self, license_key: str, *, auth: bool = True) -> requests.Response: + return self.request("post", f"/license/{license_key}/enable", auth=auth) + + def update_expiration( + self, + license_key: str, + expiration_iso: Optional[str], + *, + auth: bool = True, + ) -> requests.Response: + params = {"expiration_date": expiration_iso} if expiration_iso is not None else None + return self.request("post", f"/license/{license_key}/expiration", params=params, auth=auth) + + def export_licenses(self, *, auth: bool = True) -> requests.Response: + return self.request("get", "/license/export", auth=auth) + + def export_history(self, token: Optional[str] = None, *, auth: bool = True) -> requests.Response: + params = {"token": token} if token else None + return self.request("get", "/history/export", params=params, auth=auth) + + +@pytest.fixture(scope="session") +def base_url() -> str: + return _resolve_base_url() + + +@pytest.fixture(scope="session") +def api_key() -> str: + return os.getenv("LICENSE_SERVER_API_KEY", "API_CHANGEME") + + +@pytest.fixture(scope="session") +def client(base_url: str, api_key: str) -> LicenseServerClient: + resolved_timeout = _resolve_timeout() + license_client = LicenseServerClient(base_url, api_key, timeout=resolved_timeout) + yield license_client + license_client.close() + + +def test_server_info_endpoint(client: LicenseServerClient) -> None: + response = client.server_info() + assert response.status_code == 200 + payload = response.json() + assert "version" in payload + assert "Micro License Server" in payload["version"] + + +def test_license_endpoint_requires_authentication(client: LicenseServerClient) -> None: + response = client.create_license(auth=False) + assert response.status_code == 401 + + license_export = client.export_licenses(auth=False) + assert license_export.status_code == 401 + + history_export = client.export_history(auth=False) + assert history_export.status_code == 401 + + +def test_license_lifecycle_and_exports(client: LicenseServerClient) -> None: + create_response = client.create_license() + assert create_response.status_code == 201 + created_payload = create_response.json() + license_key = created_payload["license_key"] + assert created_payload["is_active"] is True + parts = license_key.split("-") + assert len(parts) == 5 + assert all(len(part) == 5 for part in parts) + + validity_response = client.is_valid(license_key) + assert validity_response.status_code == 200 + assert validity_response.json() is True + + disable_response = client.disable_license(license_key) + assert disable_response.status_code == 200 + assert disable_response.json()["is_active"] is False + + after_disable = client.is_valid(license_key) + assert after_disable.status_code == 200 + assert after_disable.json() is False + + enable_response = client.enable_license(license_key) + assert enable_response.status_code == 200 + assert enable_response.json()["is_active"] is True + + future_expiration = (datetime.now(timezone.utc) + timedelta(hours=1)).replace(microsecond=0) + expiration_response = client.update_expiration(license_key, future_expiration.isoformat()) + assert expiration_response.status_code == 200 + expiration_payload = expiration_response.json() + returned_expiration = datetime.fromisoformat(expiration_payload["expiration_timestamp"]) + assert abs((returned_expiration - future_expiration).total_seconds()) <= 1 + + clear_expiration_response = client.update_expiration(license_key, None) + assert clear_expiration_response.status_code == 200 + assert clear_expiration_response.json()["expiration_timestamp"] is None + + export_response = client.export_licenses() + assert export_response.status_code == 200 + export_rows = list(csv.DictReader(export_response.text.splitlines())) + matching_rows = [row for row in export_rows if row["license_key"] == license_key] + assert matching_rows, "Created license key not found in export." + assert matching_rows[0]["is_active"] == "true" + + history_response = client.export_history(token=license_key) + assert history_response.status_code == 200 + history_rows = list(csv.DictReader(history_response.text.splitlines())) + assert history_rows, "Expected history entries for created license key." + assert any(license_key in row["action"] for row in history_rows), "History does not reference the license key."