ksef.pyNew file, 21 lines
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()
ksef.pyChanged in the file: 3 lines
1challenge = requests.post(f"{API}/auth/challenge", timeout=30).json()23
ksef.pyChanged in the file: 12 lines
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() 13operation = init.json()14auth = {"Authorization": f"Bearer {operation['authenticationToken']['token']}"}
ksef.pyChanged in the file: 12 lines
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) 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"]
ksef.pyChanged in the file: 9 lines
1key, iv = os.urandom(32), os.urandom(16) 2 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() 11session_ref = opened.json()["referenceNumber"]
ksef.pyChanged in the file: 21 lines
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() 27invoice_ref = sent.json()["referenceNumber"]
ksef.pyChanged in the file: 10 lines
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: 9 print("Już przyjęta:", invoice["status"]["extensions"]["originalKsefNumber"])10 break11 if code not in (100, 150):12 raise RuntimeError(invoice["status"]) 13 time.sleep(2)
ksef.pyChanged in the file: 9 lines
1requests.post(f"{API}/sessions/online/{session_ref}/close",2 headers=access, timeout=30).raise_for_status() 3 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)