Jak wysłać FV do KSeF?

Witajcie,

Mam pytanko próbuję zapisać FV na KSeF i utknąłem

import requests
import json
import base64
import hashlib

# Zmienne wejściowe
url = 'https://ksef-test.mf.gov.pl/api/online/Invoice/Send'
session_token = '<<TOKEN_DO_TESTOWKI>>'

# Wczytanie zawartości pliku XML
with open('FV.xml', 'r', encoding='utf-8') as file:
    faktura2 = file.read()

# Generowanie danych JSON
invoice_hash = {
    "fileSize": len(faktura2),
    "hashSHA": {
        "algorithm": "SHA-256",
        "encoding": "Base64",
        "value": base64.b64encode(hashlib.sha256(faktura2.encode('utf-8')).digest()).decode('utf-8')
    }
}

invoice_payload = {
    "invoiceBody": base64.b64encode(faktura2.encode('utf-8')).decode('utf-8'),
    "type": 'plain'
}

data = {
    "invoiceHash": invoice_hash,
    "invoicePayload": invoice_payload
}

headers = {
    'accept': 'application/json',
    'SessionToken': session_token,
    'Content-Type': 'application/json'
}

# Wysyłanie żądania HTTP PUT
response = requests.put(url, data=json.dumps(data), headers=headers)

# Przetwarzanie odpowiedzi
print(response.status_code)
print(response.json())

ciągle zwraca mi taki kod błędu:

400
{'exception': {'serviceCtx': 'default', 'serviceCode': '20231120-EX-4DF5E579F7-BBF24BD286-22', 'serviceName': 'online.invoice.invoice.send', 'timestamp': '2023-11-20T19:43:02.122Z', 'exceptionDetailList': [{'exceptionCode': 21149, 'exceptionDescription': 'Brak sesji.'}]}}

FV na bank poprawna ale nie do końca kumam co robię źle. Podpowiecie?

Token (ten który zapewne wygenerowałeś w aplikacji KSeF) i session token to dwie różne rzeczy. Tokenu używasz żeby przy pomocy wyzwania autoryzacyjnego (/online/Session/AuthorisationChallenge) otworzyc sesję interaktywną (/online/Session/InitToken) i dostac session token. Wtedy możesz wysyłac faktury.

s = requests.session()
s.headers += headers

response = s.put(url, data=json.dumps(data))

Tak mi przyszło do głowy xd może zadziała

Dobra coś tam mi poszło ale utknąłem na:

przykładowe żądanie z:
https://ksef-test.mf.gov.pl/document/InitSessionTokenRequestExample/1.0

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<ns3:InitSessionTokenRequest
	xmlns="http://ksef.mf.gov.pl/schema/gtw/svc/online/types/2021/10/01/0001"
	xmlns:ns2="http://ksef.mf.gov.pl/schema/gtw/svc/types/2021/10/01/0001"
	xmlns:ns3="http://ksef.mf.gov.pl/schema/gtw/svc/online/auth/request/2021/10/01/0001">
	<ns3:Context>
		<Challenge>TUTAJ PRZEKAZUJE CHALLENGE z /online/Session/AuthorisationChallenge i to już mam</Challenge>
		<Identifier xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="ns2:SubjectIdentifierByCompanyType">
			<ns2:Identifier>TUTAJ NIP</ns2:Identifier>
		</Identifier>
		<DocumentType>
			<ns2:Service>KSeF</ns2:Service>
			<ns2:FormCode>
				<ns2:SystemCode>FA (1)</ns2:SystemCode>
				<ns2:SchemaVersion>1-0E</ns2:SchemaVersion>
				<ns2:TargetNamespace>http://crd.gov.pl/wzor/2021/11/29/11089/</ns2:TargetNamespace>
				<ns2:Value>FA</ns2:Value>
			</ns2:FormCode>
		</DocumentType>
		<Encryption>
			<ns2:EncryptionKey>
				<ns2:Encoding>Base64</ns2:Encoding>
				<ns2:Algorithm>AES</ns2:Algorithm>
				<ns2:Size>256</ns2:Size>				<ns2:Value>Kmrs/g9ZjI5lKDoMukGDMDULgvx0BkZ6qHEazLLo+LmzK4FNhNp4owcPioydkuiJE02norz5P5hjFO+SMO9kgtPji0Qjx57B+o6tKMUlL6EA3nn40+JNkZuvW+hUnEt7/kxZCcjxEXTuLpQuANwovUxdej041nKRY+sthBEDb3gN4pzJ+HfACysB8H5dpvZj1MkBe7hydLCJlNkEYHYA5dO/3FnW/5fh12Be3ij7Z410+3XZT5AiteEwGCJCPdC0epM3yJgi8uZt2vqn83fgYqSqmMtXLk0ZVjTT4Z1wd2cZ3DF7tlIfoYeB2DKQiwbXAZr/W3jVzYliXBykZLC/Sg==</ns2:Value>
			</ns2:EncryptionKey>
			<ns2:EncryptionInitializationVector>
				<ns2:Encoding>Base64</ns2:Encoding>
				<ns2:Bytes>16</ns2:Bytes>
				<ns2:Value>H9KQyy/SHurlswE4pxSIsg==</ns2:Value>
			</ns2:EncryptionInitializationVector>
			<ns2:EncryptionAlgorithmKey>
				<ns2:Algorithm>RSA</ns2:Algorithm>
				<ns2:Mode>ECB</ns2:Mode>
				<ns2:Padding>PKCS#1</ns2:Padding>
			</ns2:EncryptionAlgorithmKey>
			<ns2:EncryptionAlgorithmData>
				<ns2:Algorithm>AES</ns2:Algorithm>
				<ns2:Mode>CBC</ns2:Mode>
				<ns2:Padding>PKCS#7</ns2:Padding>
			</ns2:EncryptionAlgorithmData>
		</Encryption>
<Token>uH6dmA9lr7/5izGMOTXVzIAe1awXZYZ61zLCEaImSLP+BftDtziV5I+4EaKvWZ+IUCkXHP8FEGDsLHfefUVLS5vbZ8r0W5UxiSqqOAKxvkE7pCfh31+AmKZxZa0TlXYK5EYa4RkJhm7HTgKe4WGZ/Y4G2PdCzdImtvZL49yqQQ3bLqeGeVJ9rMzkuxCbtKGbdopN1/V+64fvLClIuolWI+/z0FlxDVE9a0f8EFKPPpzGowVwJcn5PHBkvVh45vT+pyNbDSvEsyL/udRBXCioWbgRaZReRJNIg6YWZLTSHyi8BK4ofQVuhdcPwHjkETmEQ0aktj+uyqmxqvHbFTA93Q==</Token>
	</ns3:Context>
</ns3:InitSessionTokenRequest>

Natomiast nie ogarniam TOKEN'a (to jest ten TOKEN API ? ) oraz sekcji Encryption (mam sobie klucz wygenerować lokalnie i co z nim zrobić?) podpowie ktoś coś?

Dobra ktoś mi podpowie co zrobiłem źle z tym tokenem?

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from datetime import datetime
import base64
from dateutil import parser

def encrypt_with_ksef_key(ksef_server_key, to_encrypt):
    try:
        # Sprawdź, czy klucz jest w formacie PEM czy DER
        if b'-----BEGIN PUBLIC KEY-----' in ksef_server_key:
            public_key = serialization.load_pem_public_key(ksef_server_key, backend=default_backend())
        else:
            public_key = serialization.load_der_public_key(ksef_server_key, backend=default_backend())

        encrypted = public_key.encrypt(
            to_encrypt,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )

        return base64.b64encode(encrypted).decode('utf-8')
    except Exception as e:
        raise RuntimeError(e)


def get_auth_token(timestamp, properties):
    auth_token = f"{properties['token']}|{int(parser.parse(timestamp).timestamp())}"
    return encrypt_with_ksef_key(properties['klucz_serwera'], auth_token.encode('utf-8'))

def load_and_encode_server_key(file_path):
    try:
        with open(file_path, 'rb') as key_file:
            return key_file.read()
    except Exception as e:
        raise RuntimeError(e)

# Przykład użycia
properties = {
    'token': '4EA55DA74A1B9A92E064CC1DD8EC82CA71DBB133E6B697E5BF8E8837760B5B83',
    'klucz_serwera': load_and_encode_server_key('publicKey.pem')
}

def get_finall_auth_token(ptimestmap):
    return get_auth_token(ptimestmap, properties)

bo dostaj:

Poprawna odpowiedź. Timestamp: 2023-12-14T20:26:01.382Z, Challenge: 20231214-CR-E4789BD2BE-9818FE9D91-B6
Błąd: {'serviceCtx': 'srvTEMFB', 'serviceCode': '20231214-EX-C868420B97-BCC9EB0D3B-15', 'serviceName': 'online.session.session.token.init', 'timestamp': '2023-12-14T20:26:01.543Z', 'referenceNumber': '20231214-SE-4C7760140E-D48AD14A0B-2F', 'exceptionDetailList': [{'exceptionCode': 20005, 'exceptionDescription': 'Nieprawidłowo zaszyfrowana treść.'}]}

def encrypt_with_public_key_pem_and_base64(data, public_key_path):
    with open(public_key_path, 'rb') as key_file:
        public_key = serialization.load_pem_public_key(
            key_file.read(),
            backend=default_backend()
        )

    # Zakodowanie danych
    encrypted_data = public_key.encrypt(
        data.encode('utf-8'),
        asymmetric.padding.PKCS1v15()
    )

    # Zakodowanie zaszyfrowanych danych do Base64
    encrypted_data_base64 = base64.b64encode(encrypted_data).decode('utf-8')

    return encrypted_data_base64

oraz

data_to_encrypt = auth_token+'|'+str(timestamp_unix_ms)
public_key_path = 'publicKey.pem'  

encrypted_data = encrypt_with_public_key_pem_and_base64(data_to_encrypt, public_key_path)

Nie wiem czy nadal masz problem jeśli tak to, ten kod działa