KSeF API 2.0 krok po kroku: wysyłka faktury z własnego systemu

1import base642import requests3from cryptography import x5094from cryptography.hazmat.primitives import hashes5from cryptography.hazmat.primitives.asymmetric import padding6 7API = "https://api-test.ksef.mf.gov.pl/v2"8 9 10def mf_key(usage: str):11    certs = requests.get(f"{API}/security/public-key-certificates", timeout=30).json()12    cert = next(c for c in certs if usage in c["usage"])13    der = base64.b64decode(cert["certificate"])14    return x509.load_der_x509_certificate(der).public_key()15 16 17def rsa_oaep(key, data: bytes) -> str:18    sealed = key.encrypt(data, padding.OAEP(19        mgf=padding.MGF1(algorithm=hashes.SHA256()),20        algorithm=hashes.SHA256(), label=None))21    return base64.b64encode(sealed).decode()

1challenge = requests.post(f"{API}/auth/challenge", timeout=30).json()2# {"challenge": "20261006-CR-...", "timestamp": "2026-10-06T12:33:47+00:00",3#  "timestampMs": 1791290027137, "clientIp": "..."}

1import os2 3KSEF_TOKEN = os.environ["KSEF_TOKEN"]4NIP = os.environ["KSEF_NIP"]5 6plain = f"{KSEF_TOKEN}|{challenge['timestampMs']}".encode()7init = requests.post(f"{API}/auth/ksef-token", json={8    "challenge": challenge["challenge"],9    "contextIdentifier": {"type": "Nip", "value": NIP},10    "encryptedToken": rsa_oaep(mf_key("KsefTokenEncryption"), plain),11}, timeout=30)12init.raise_for_status()  # 202 Accepted13operation = init.json()14auth = {"Authorization": f"Bearer {operation['authenticationToken']['token']}"}

1import time2 3while True:4    status = requests.get(f"{API}/auth/{operation['referenceNumber']}",5                          headers=auth, timeout=30).json()["status"]6    if status["code"] == 200:7        break8    if status["code"] != 100:9        raise RuntimeError(status)  # 415, 425, 450...10    time.sleep(1)11 12tokens = requests.post(f"{API}/auth/token/redeem", headers=auth, timeout=30).json()13access = {"Authorization": f"Bearer {tokens['accessToken']['token']}"}14refresh_token = tokens["refreshToken"]["token"]

1key, iv = os.urandom(32), os.urandom(16)  # AES-256 key and IV for this session2 3opened = requests.post(f"{API}/sessions/online", headers=access, json={4    "formCode": {"systemCode": "FA (3)", "schemaVersion": "1-0E", "value": "FA"},5    "encryption": {6        "encryptedSymmetricKey": rsa_oaep(mf_key("SymmetricKeyEncryption"), key),7        "initializationVector": base64.b64encode(iv).decode(),8    },9}, timeout=30)10opened.raise_for_status()  # 201 Created11session_ref = opened.json()["referenceNumber"]

1import hashlib2from cryptography.hazmat.primitives import padding as sym_padding3from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes4 5 6def aes_cbc(data: bytes) -> bytes:7    padder = sym_padding.PKCS7(128).padder()8    padded = padder.update(data) + padder.finalize()9    encryptor = Cipher(algorithms.AES(key), modes.CBC(iv)).encryptor()10    return encryptor.update(padded) + encryptor.finalize()11 12 13def sha256_b64(data: bytes) -> str:14    return base64.b64encode(hashlib.sha256(data).digest()).decode()15 16 17xml = open("faktura.xml", "rb").read()18encrypted = aes_cbc(xml)19sent = requests.post(f"{API}/sessions/online/{session_ref}/invoices", headers=access, json={20    "invoiceHash": sha256_b64(xml),21    "invoiceSize": len(xml),22    "encryptedInvoiceHash": sha256_b64(encrypted),23    "encryptedInvoiceSize": len(encrypted),24    "encryptedInvoiceContent": base64.b64encode(encrypted).decode(),25}, timeout=60)26sent.raise_for_status()  # 202 Accepted27invoice_ref = sent.json()["referenceNumber"]

1while True:2    invoice = requests.get(f"{API}/sessions/{session_ref}/invoices/{invoice_ref}",3                           headers=access, timeout=30).json()4    code = invoice["status"]["code"]5    if code == 200:6        print("Numer KSeF:", invoice["ksefNumber"])7        break8    if code == 440:  # duplicate: the same file was accepted before9        print("Już przyjęta:", invoice["status"]["extensions"]["originalKsefNumber"])10        break11    if code not in (100, 150):12        raise RuntimeError(invoice["status"])  # 430, 435, 450...13    time.sleep(2)

1requests.post(f"{API}/sessions/online/{session_ref}/close",2              headers=access, timeout=30).raise_for_status()  # 2043 4while True:5    session = requests.get(f"{API}/sessions/{session_ref}", headers=access, timeout=30).json()6    pages = (session.get("upo") or {}).get("pages") or []7    if pages:8        for page in pages:9            upo = requests.get(page["downloadUrl"], timeout=60)10            open(f"upo-{page['referenceNumber']}.xml", "wb").write(upo.content)11        break12    time.sleep(5)