"""Sign files on the Muhrdor server with a service account (ADR-0037, ADR-0055). Python 3.10+ standard library only - copy this file into your system and adapt it. export MUHRDOR_CLIENT_ID=sa_... # panel -> "API kalitlari" export MUHRDOR_CLIENT_SECRET=... # shown once when the account is created export MUHRDOR_BASE=https://muhrdor.uz # optional python muhrdor_sign.py --site my.soliq.uz --credential report1.xml ... For every file ``.p7b`` (PKCS#7, DER) is written next to it and a line ``: `` is printed. The key owner must first give the service account a grant act naming the site; ``--credential`` is a ``credential_id`` from ``GET /api/v1/client/credentials``. Exit code: 0 - every file signed; 1 - at least one was refused or failed; 2 - usage. What it shows: one token per run (an hour is plenty), batches of at most 200 documents and 4 MiB, an ``Idempotency-Key`` per batch that is *kept* when the batch is retried after a lost answer (the server then returns the first answer instead of signing twice), and waiting as long as ``Retry-After`` says. """ from __future__ import annotations import argparse import base64 import json import os import re import sys import time import urllib.error import urllib.request import uuid from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Any, TextIO MAX_DOCUMENTS = 200 # per request (the server's limit) MAX_BATCH_BYTES = 4 * 1024 * 1024 # the server takes 5 MiB of documents per request TIMEOUT = 120.0 # seconds; a full batch is 200 HSM operations RATE_LIMIT_TRIES = 3 Opener = Callable[..., Any] class ApiFailure(Exception): def __init__( self, status: int, code: str, reason: str | None, retry_after: str | None = None, retryable: bool = False, ) -> None: super().__init__(f"{status} {code}{'' if reason is None else f' ({reason})'}") self.status = status self.code = code self.reason = reason self.retry_after = retry_after # seconds, from the Retry-After header (429) self.retryable = retryable # signing_failed: the same request may succeed def external_ref(name: str) -> str: """The file name as the server accepts a reference: [A-Za-z0-9._:/-], at most 64.""" return re.sub(r"[^A-Za-z0-9._:/-]+", "_", name)[:64] or "document" class Client: def __init__( self, base: str, client_id: str, secret: str, opener: Opener, sleep: Callable[[float], None] ) -> None: if not (base.startswith("https://") or base.startswith("http://localhost")): raise ValueError("MUHRDOR_BASE must be an https:// address") self.base = base.rstrip("/") self.client_id = client_id self.secret = secret self.opener = opener self.sleep = sleep self.token: str | None = None def _post(self, path: str, body: dict[str, Any], headers: Mapping[str, str]) -> dict[str, Any]: request = urllib.request.Request( # noqa: S310 (the scheme is checked in __init__) self.base + path, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json", **headers}, method="POST", ) try: with self.opener(request, timeout=TIMEOUT) as response: answer: dict[str, Any] = json.loads(response.read()) return answer except urllib.error.HTTPError as exc: try: error = json.loads(exc.read()).get("error", {}) except (ValueError, AttributeError): error = {} raise ApiFailure( exc.code, str(error.get("code", "http_error")), error.get("reason"), exc.headers.get("Retry-After") if exc.headers else None, error.get("retryable") is True, ) from exc def _new_token(self) -> str: answer = self._post( "/api/v1/token", { "grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.secret, }, {}, ) self.token = str(answer["access_token"]) return self.token def sign( self, credential: str, site: str, documents: list[dict[str, Any]] ) -> list[dict[str, Any]]: """One batch. The key is chosen once and reused for every retry of this batch.""" key = f"batch-{uuid.uuid4().hex}" body = {"credential_id": credential, "site": site, "documents": documents} network_retry = token_retry = retryable_retry = True limited = 0 while True: token = self.token or self._new_token() headers = {"Authorization": f"Bearer {token}", "Idempotency-Key": key} try: results: list[dict[str, Any]] = self._post( "/api/v1/signatures/sign", body, headers )["results"] return results except urllib.error.URLError: if not network_retry: raise network_retry = False # the answer may be lost, not the signature: same key self.sleep(2.0) except ApiFailure as failure: if failure.status == 401 and token_retry: token_retry = False self.token = None # expired: a new one, same key elif failure.status == 429 and limited < RATE_LIMIT_TRIES: limited += 1 self.sleep(float(failure.retry_after or 60)) elif failure.code == "request_in_progress" and limited < RATE_LIMIT_TRIES: limited += 1 self.sleep(1.0) elif failure.code == "signing_failed" and failure.retryable and retryable_retry: retryable_retry = False self.sleep(2.0) else: raise def batches(paths: Sequence[Path]) -> list[list[Path]]: out: list[list[Path]] = [] size = 0 for path in paths: length = path.stat().st_size if not out or len(out[-1]) >= MAX_DOCUMENTS or size + length > MAX_BATCH_BYTES: out.append([]) size = 0 out[-1].append(path) size += length return out def main( argv: Sequence[str] | None = None, *, env: Mapping[str, str] = os.environ, opener: Opener = urllib.request.urlopen, sleep: Callable[[float], None] = time.sleep, out: TextIO = sys.stdout, ) -> int: parser = argparse.ArgumentParser( description="Sign files on the Muhrdor server (service account)." ) parser.add_argument( "--site", required=True, help="the site the documents are for, e.g. my.soliq.uz" ) parser.add_argument( "--credential", required=True, help="credential_id (GET /api/v1/client/credentials)" ) parser.add_argument("--attached", action="store_true", help="embed the content in the PKCS#7") parser.add_argument("files", nargs="+", type=Path) args = parser.parse_args(argv) client_id = env.get("MUHRDOR_CLIENT_ID", "") secret = env.get("MUHRDOR_CLIENT_SECRET", "") if client_id == "" or secret == "": out.write("set MUHRDOR_CLIENT_ID and MUHRDOR_CLIENT_SECRET\n") return 2 try: client = Client( env.get("MUHRDOR_BASE", "https://muhrdor.uz"), client_id, secret, opener, sleep ) except ValueError as exc: out.write(f"{exc}\n") return 2 failed = 0 for batch in batches(args.files): documents = [ { "content_b64": base64.b64encode(path.read_bytes()).decode("ascii"), "detached": not args.attached, "external_ref": external_ref(path.name), } for path in batch ] try: results = client.sign(args.credential, args.site, documents) except (ApiFailure, urllib.error.URLError) as exc: for path in batch: out.write(f"{path}: FAILED {exc}\n") failed += len(batch) continue for result in results: path = batch[int(result["index"])] error = result.get("error") if error: reason = error.get("reason") out.write( f"{path}: REFUSED {error.get('code')}{'' if reason is None else f' {reason}'}\n" ) failed += 1 continue Path(f"{path}.p7b").write_bytes(base64.b64decode(result["pkcs7_b64"])) out.write(f"{path}: {result['audit_ref']}\n") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())