Deactivate an installation

{}

{
  "success": true,
  "deactivated": true,
  "request_id": "req_29039d899dd3655d7ada1c89"
}

"""Inspect one Sandbox API operation; transport example, not access authorization.
Requires Python3.9+ standard library. Run: python3 request.py public-key
Expected: full API JSON on stdout; exit0 on success,1 on API rejection,2 on setup/transport failure.
Credentials/body via KDP_* environment variables. Stdout may contain new secrets:
use only a private terminal, never application/CI logs. Verify signed state separately.
"""
import json
import os
import re
import sys
from urllib.error import HTTPError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

# Preserve raw UTF-8 JSON even when piped on a non-UTF-8 Windows console.
sys.stdout.reconfigure(encoding='utf-8')

OPERATIONS = json.loads(r'''{
  "public-key": {
    "method": "GET",
    "path": "/public-key",
    "auth": "public"
  },
  "create-license": {
    "method": "POST",
    "path": "/licenses",
    "auth": "developer"
  },
  "activate": {
    "method": "POST",
    "path": "/licenses/activate",
    "auth": "publicApp"
  },
  "validate": {
    "method": "POST",
    "path": "/licenses/validate",
    "auth": "activation"
  },
  "deactivate": {
    "method": "POST",
    "path": "/licenses/deactivate",
    "auth": "activation"
  },
  "get-license": {
    "method": "GET",
    "path": "/licenses/{license}",
    "auth": "developer"
  },
  "revoke": {
    "method": "POST",
    "path": "/licenses/{license}/revoke",
    "auth": "developer"
  },
  "suspend": {
    "method": "POST",
    "path": "/licenses/{license}/suspend",
    "auth": "developer"
  },
  "restore": {
    "method": "POST",
    "path": "/licenses/{license}/restore",
    "auth": "developer"
  },
  "renew": {
    "method": "POST",
    "path": "/licenses/{license}/renew",
    "auth": "developer"
  },
  "reset-devices": {
    "method": "POST",
    "path": "/licenses/{license}/reset-devices",
    "auth": "developer"
  }
}''')


def required(name):
    value = os.environ.get(name, '')
    if not value or '\r' in value or '\n' in value:
        raise ValueError('Set a valid ' + name + '.')
    return value


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None


def main():
    if len(sys.argv) != 2 or sys.argv[1] not in OPERATIONS:
        raise ValueError('Choose an operation: ' + ', '.join(OPERATIONS))
    operation = OPERATIONS[sys.argv[1]]
    base = os.environ.get('KDP_BASE_URL', 'https://licensing.kdpika.com/api/v1').rstrip('/')
    parsed = urlsplit(base)
    loopback = parsed.scheme == 'http' and parsed.hostname in ('127.0.0.1', 'localhost') and os.environ.get('KDP_ALLOW_LOOPBACK') == '1'
    if not parsed.hostname or (parsed.scheme != 'https' and not loopback) or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path != '/api/v1':
        raise ValueError('Use an HTTPS API base URL ending in /api/v1. Loopback HTTP needs explicit test opt-in.')
    path = operation['path']
    if '{license}' in path:
        identifier = required('KDP_LICENSE_ID')
        if not re.fullmatch(r'[A-Za-z0-9_-]+', identifier):
            raise ValueError('This sample accepts the simple license IDs issued by the platform.')
        path = path.replace('{license}', identifier)
    headers = {'Accept': 'application/json', 'User-Agent': 'KDPikaDocsExample/1.0 Python'}
    if operation['auth'] == 'developer':
        key = required('KDP_API_KEY')
        if not key.startswith('lmp_sandbox_'):
            raise ValueError('Use a Sandbox developer API key with this example.')
        headers['Authorization'] = 'Bearer ' + key
    elif operation['auth'] == 'activation':
        headers['Authorization'] = 'Bearer ' + required('KDP_ACTIVATION_TOKEN')
    elif operation['auth'] == 'publicApp':
        headers.update({'X-App-ID': required('KDP_APP_ID'), 'X-Environment': 'SANDBOX'})
    if sys.argv[1] == 'create-license':
        headers['Idempotency-Key'] = required('KDP_IDEMPOTENCY_KEY')
    if sys.argv[1] == 'validate':
        headers['X-Installation-ID'] = required('KDP_INSTALLATION_ID')
    body = None
    if operation['method'] == 'POST':
        source = os.environ.get('KDP_BODY_JSON', '{}')
        if not isinstance(json.loads(source), dict):
            raise ValueError('KDP_BODY_JSON must be a JSON object.')
        body = source.encode('utf-8')
        headers['Content-Type'] = 'application/json'
    request = Request(base + path, data=body, headers=headers, method=operation['method'])
    try:
        response = build_opener(NoRedirect()).open(request, timeout=15)
    except HTTPError as error:
        response = error
    raw = response.read().decode('utf-8')
    result = json.loads(raw)
    sys.stdout.write(raw + '\n')
    if not (200 <= response.status < 300 and result.get('success') is True):
        sys.stderr.write('The API rejected this request. Check error.code and request_id; do not blindly retry a mutation.\n')
        return 1
    return 0


if __name__ == '__main__':
    try:
        sys.exit(main())
    except ValueError as error:
        # Configuration errors never interpolate credential values.
        sys.stderr.write(str(error) if str(error).startswith(('Set a valid KDP_', 'Choose an operation:', 'Use ', 'This sample ', 'KDP_BODY_JSON ')) else 'The response or request was not valid JSON.')
        sys.stderr.write('\n')
        sys.exit(2)
    except Exception:
        sys.stderr.write('Request unavailable. Check connectivity; reconcile uncertain mutations before retrying.\n')
        sys.exit(2)