mi
73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
import base64
|
|
import os
|
|
import uuid
|
|
|
|
os.environ.setdefault("BITRIX_DATABASE_URL", "postgresql://unused/unused")
|
|
os.environ.setdefault("BITRIX_CLIENT_ID", "client")
|
|
os.environ.setdefault("BITRIX_CLIENT_SECRET", "secret")
|
|
os.environ.setdefault("BITRIX_APPLICATION_TOKEN", "application-token")
|
|
os.environ.setdefault("BITRIX_INTERNAL_API_TOKEN", "internal-token-32-characters-long")
|
|
os.environ.setdefault("BITRIX_API_FORWARD_URL", "http://api/internal/openlines/v1/inbox")
|
|
os.environ.setdefault("BITRIX_API_FORWARD_TOKEN", "forward-token-32-characters-long")
|
|
os.environ.setdefault(
|
|
"BITRIX_TOKEN_ENCRYPTION_KEY",
|
|
base64.urlsafe_b64encode(b"k" * 32).decode().rstrip("="),
|
|
)
|
|
|
|
import pytest
|
|
|
|
from app.main import (
|
|
TokenCipher,
|
|
canonical_fingerprint,
|
|
normalize_event,
|
|
retry_delay,
|
|
validate_portal,
|
|
)
|
|
|
|
|
|
def test_token_cipher_binds_aad():
|
|
cipher = TokenCipher(os.environ["BITRIX_TOKEN_ENCRYPTION_KEY"], "v1")
|
|
ciphertext, nonce = cipher.encrypt("secret", "member", "han0107.bitrix24.ru", "access")
|
|
assert cipher.decrypt(ciphertext, nonce, "member", "han0107.bitrix24.ru", "access") == "secret"
|
|
with pytest.raises(Exception):
|
|
cipher.decrypt(ciphertext, nonce, "other", "han0107.bitrix24.ru", "access")
|
|
|
|
|
|
def test_normalize_message_and_finish():
|
|
external = str(uuid.uuid4())
|
|
message = normalize_event(
|
|
{
|
|
"event": "ONIMCONNECTORMESSAGEADD",
|
|
"data": {
|
|
"MESSAGES": [
|
|
{
|
|
"im": {"chat_id": 1807, "message_id": 86497},
|
|
"chat": {"id": external},
|
|
"message": {"text": "Ответ", "files": []},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
assert message["event_type"] == "message.new"
|
|
assert message["external_chat_id"] == external
|
|
assert message["bitrix_message_id"] == "86497"
|
|
closed = normalize_event(
|
|
{"event": "ONIMCONNECTORDIALOGFINISH", "data": {"external_chat_id": external}}
|
|
)
|
|
assert closed["event_type"] == "dialog.closed"
|
|
|
|
|
|
def test_fingerprint_ignores_signed_query_and_portal_validation():
|
|
payload = {"message": {"files": [{"download_url": "https://s3/object?sig=one"}]}}
|
|
other = {"message": {"files": [{"download_url": "https://s3/object?sig=two"}]}}
|
|
assert canonical_fingerprint(payload) == canonical_fingerprint(other)
|
|
validate_portal(
|
|
"han0107.bitrix24.ru",
|
|
"https://han0107.bitrix24.ru/rest/",
|
|
"han0107.bitrix24.ru",
|
|
)
|
|
with pytest.raises(ValueError):
|
|
validate_portal("evil.example", "https://evil.example/rest/", "han0107.bitrix24.ru")
|
|
assert 0 <= retry_delay(4, 300) <= 8
|