diff --git a/python/src/vaas/messages/file_report.py b/python/src/vaas/messages/file_report.py index 3f265266..d59a24f5 100644 --- a/python/src/vaas/messages/file_report.py +++ b/python/src/vaas/messages/file_report.py @@ -7,3 +7,4 @@ class FileReport(BaseModel): detection: Optional[str] fileType: Optional[str] mimeType: Optional[str] + isEncrypted: Optional[bool] = None diff --git a/python/src/vaas/messages/url_report.py b/python/src/vaas/messages/url_report.py index b4986913..1ef98659 100644 --- a/python/src/vaas/messages/url_report.py +++ b/python/src/vaas/messages/url_report.py @@ -7,4 +7,5 @@ class UrlReport(BaseModel): url: str detection: Optional[str] fileType: Optional[str] - mimeType: Optional[str] \ No newline at end of file + mimeType: Optional[str] + isEncrypted: Optional[bool] = None \ No newline at end of file diff --git a/python/src/vaas/messages/vaas_verdict.py b/python/src/vaas/messages/vaas_verdict.py index e5150d21..3b389c40 100644 --- a/python/src/vaas/messages/vaas_verdict.py +++ b/python/src/vaas/messages/vaas_verdict.py @@ -7,6 +7,7 @@ class VaasVerdict(BaseModel): detection: Optional[str] fileType: Optional[str] mimeType: Optional[str] + isEncrypted: Optional[bool] = None @staticmethod def from_report(report): @@ -15,5 +16,6 @@ def from_report(report): verdict=report.verdict, detection=report.detection, fileType=report.fileType, - mimeType=report.mimeType + mimeType=report.mimeType, + isEncrypted=report.isEncrypted ) diff --git a/python/tests/test_vaas.py b/python/tests/test_vaas.py index 0d849bb2..34b4b2cb 100644 --- a/python/tests/test_vaas.py +++ b/python/tests/test_vaas.py @@ -24,6 +24,10 @@ EICAR_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/eicar.com.txt" CLEAN_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" PUP_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/PotentiallyUnwanted.exe" +PASSWORD_ZIP_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/password.zip" +WITH_AND_WITHOUT_PASSWORD_ZIP_URL = ( + "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/with-and-without-password.zip" +) CLEAN_FILE_CONTENT = "I am clean." @@ -32,18 +36,14 @@ class TestVaas: @pytest.mark.asyncio() @pytest.mark.parametrize( "sha256, expected_verdict", - [ - (EICAR_SHA256, "Malicious"), - (CLEAN_SHA256, "Clean"), - (PUP_SHA256, "Pup") - ], - ids=["Malware", "Clean", "Pup"] + [(EICAR_SHA256, "Malicious"), (CLEAN_SHA256, "Clean"), (PUP_SHA256, "Pup")], + ids=["Malware", "Clean", "Pup"], ) async def test_for_sha256_returns_verdict(self, vaas, sha256, expected_verdict): - verdict = await vaas.for_sha256(sha256) + verdict = await vaas.for_sha256(sha256) - assert verdict.verdict == expected_verdict - assert verdict.sha256.casefold() == sha256.casefold() + assert verdict.verdict == expected_verdict + assert verdict.sha256.casefold() == sha256.casefold() @pytest.mark.asyncio() @pytest.mark.parametrize( @@ -54,9 +54,11 @@ async def test_for_sha256_returns_verdict(self, vaas, sha256, expected_verdict): (True, False), (True, True), ], - ids=["false_for_all", "only_hash_lookup", "only_cache", "true_for_all"] + ids=["false_for_all", "only_hash_lookup", "only_cache", "true_for_all"], ) - async def test_for_sha256_send_options(self, vaas, use_cache, use_hash_lookup, httpx_mock): + async def test_for_sha256_send_options( + self, vaas, use_cache, use_hash_lookup, httpx_mock + ): request_url = f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(use_cache).lower()}&useHashLookup={str(use_hash_lookup).lower()}" httpx_mock.add_response( method="GET", @@ -67,8 +69,8 @@ async def test_for_sha256_send_options(self, vaas, use_cache, use_hash_lookup, h "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -78,7 +80,9 @@ async def test_for_sha256_send_options(self, vaas, use_cache, use_hash_lookup, h actual_request = httpx_mock.get_requests()[0] actual_url = str(actual_request.url) assert len(httpx_mock.get_requests()) == 1 - assert actual_url == request_url, f"URL mismatch:\nExpected: {request_url}\nActual: {actual_url}" + assert ( + actual_url == request_url + ), f"URL mismatch:\nExpected: {request_url}\nActual: {actual_url}" assert verdict.verdict == "Clean" @pytest.mark.asyncio() @@ -93,8 +97,8 @@ async def test_for_sha256_send_user_agent(self, vaas, httpx_mock): "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -117,8 +121,8 @@ async def test_for_sha256_set_request_id_send_trace_state(self, vaas, httpx_mock "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -131,17 +135,15 @@ async def test_for_sha256_set_request_id_send_trace_state(self, vaas, httpx_mock assert verdict.verdict == "Clean" @pytest.mark.asyncio() - async def test_for_sha256_bad_request_raise_vaas_client_error(self, vaas, httpx_mock): - request_url = (f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}" - ) + async def test_for_sha256_bad_request_raise_vaas_client_error( + self, vaas, httpx_mock + ): + request_url = f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}" httpx_mock.add_response( method="GET", url=request_url, status_code=400, - json = { - "detail": "Mocked client-side error", - "type": "VaasClientException" - } + json={"detail": "Mocked client-side error", "type": "VaasClientException"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -153,17 +155,15 @@ async def test_for_sha256_bad_request_raise_vaas_client_error(self, vaas, httpx_ assert problem_details.type == "VaasClientException" @pytest.mark.asyncio() - async def test_for_sha256_server_error_raise_vaas_server_error(self, vaas, httpx_mock): - request_url = (f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}" - ) + async def test_for_sha256_server_error_raise_vaas_server_error( + self, vaas, httpx_mock + ): + request_url = f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}" httpx_mock.add_response( method="GET", url=request_url, status_code=500, - json = { - "detail": "Mocked server-side error", - "type": "VaasServerException" - } + json={"detail": "Mocked server-side error", "type": "VaasServerException"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -175,24 +175,29 @@ async def test_for_sha256_server_error_raise_vaas_server_error(self, vaas, httpx assert problem_details.type == "VaasServerException" @pytest.mark.asyncio() - async def test_for_sha256_authentication_error_raise_vaas_authentication_error(self, vaas): - vaas.authenticator.get_token = AsyncMock(side_effect=VaasAuthenticationError("Mocked auth error")) + async def test_for_sha256_authentication_error_raise_vaas_authentication_error( + self, vaas + ): + vaas.authenticator.get_token = AsyncMock( + side_effect=VaasAuthenticationError("Mocked auth error") + ) with pytest.raises(VaasAuthenticationError): await vaas.for_sha256(CLEAN_SHA256) @pytest.mark.asyncio() - async def test_for_sha256_unauthorized_raise_vaas_authentication_error(self, vaas, httpx_mock): - request_url = (f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}" - ) + async def test_for_sha256_unauthorized_raise_vaas_authentication_error( + self, vaas, httpx_mock + ): + request_url = f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}" httpx_mock.add_response( method="GET", url=request_url, status_code=401, - json = { + json={ "detail": "Authentication error", - "type": "VaasAuthenticationException" - } + "type": "VaasAuthenticationException", + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -209,7 +214,7 @@ async def test_for_sha256_cancel_request_raise_cancel_error(self, vaas, httpx_mo httpx_mock.add_exception( method="GET", url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache=true&useHashLookup=true", - exception=asyncio.CancelledError() + exception=asyncio.CancelledError(), ) with pytest.raises(asyncio.CancelledError): @@ -217,61 +222,65 @@ async def test_for_sha256_cancel_request_raise_cancel_error(self, vaas, httpx_mo @pytest.mark.asyncio() @pytest.mark.parametrize( - "url, expected_verdict", + "url, expected_verdict, is_encrypted", [ - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/eicar.com.txt", "Malicious"), - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt", "Clean"), - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/PotentiallyUnwanted.exe", "Pup") + (EICAR_URL, "Malicious", False), + (CLEAN_URL, "Clean", False), + (PUP_URL, "Pup", False), + (PASSWORD_ZIP_URL, "Clean", True), + (WITH_AND_WITHOUT_PASSWORD_ZIP_URL, "Malicious", True), ], - ids=["Malware", "Clean", "Pup"] + ids=["Malware", "Clean", "Pup", "Encrypted", "EncryptedWithMalwareInside"], ) - async def test_for_stream_returns_verdict(self, vaas, url, expected_verdict): - async with httpx.AsyncClient() as client: - response = await client.get(url) - content_length = response.headers["Content-Length"] - verdict = await vaas.for_stream(response.aiter_bytes(), content_length) + async def test_for_stream_returns_verdict( + self, vaas, url, expected_verdict, is_encrypted + ): + async with httpx.AsyncClient() as client: + response = await client.get(url) + content_length = response.headers["Content-Length"] + verdict = await vaas.for_stream(response.aiter_bytes(), content_length) - assert verdict.verdict == expected_verdict + assert verdict.verdict == expected_verdict + assert verdict.isEncrypted == is_encrypted @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) @pytest.mark.parametrize( "use_hash_lookup", - [ - False, - True - ], - ids=["hash_lookup_enabled", "hash_lookup_disabled"] + [False, True], + ids=["hash_lookup_enabled", "hash_lookup_disabled"], ) async def test_for_stream_send_options(self, vaas, use_hash_lookup, httpx_mock): httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(use_hash_lookup).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(use_hash_lookup).lower()}", status_code=200, - json={ - "sha256": CLEAN_SHA256 - } + json={"sha256": CLEAN_SHA256}, ) httpx_mock.add_response( method="GET", - url= f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(use_hash_lookup).lower()}", + url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(use_hash_lookup).lower()}", status_code=200, json={ "sha256": CLEAN_SHA256, "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] options = ForStreamOptions(use_hash_lookup=use_hash_lookup) - verdict = await vaas.for_stream(response.aiter_bytes(), content_length, options) + verdict = await vaas.for_stream( + response.aiter_bytes(), content_length, options + ) for actual_request in httpx_mock.get_requests(): actual_url = str(actual_request.url) @@ -280,32 +289,32 @@ async def test_for_stream_send_options(self, vaas, use_hash_lookup, httpx_mock): assert verdict.verdict == "Clean" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_stream_send_user_agent(self, vaas, httpx_mock): httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=200, - json={ - "sha256": CLEAN_SHA256 - } + json={"sha256": CLEAN_SHA256}, ) httpx_mock.add_response( method="GET", - url= f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}", status_code=200, json={ "sha256": CLEAN_SHA256, "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] verdict = await vaas.for_stream(response.aiter_bytes(), content_length) for request in httpx_mock.get_requests(): @@ -315,35 +324,37 @@ async def test_for_stream_send_user_agent(self, vaas, httpx_mock): assert verdict.verdict == "Clean" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_stream_set_request_id_send_trace_state(self, vaas, httpx_mock): httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=200, - json={ - "sha256": CLEAN_SHA256 - } + json={"sha256": CLEAN_SHA256}, ) httpx_mock.add_response( method="GET", - url= f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache={str(True).lower()}&useHashLookup={str(True).lower()}", status_code=200, json={ "sha256": CLEAN_SHA256, "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] options = ForStreamOptions(vaas_request_id="foobar") - verdict = await vaas.for_stream(response.aiter_bytes(), content_length, options) + verdict = await vaas.for_stream( + response.aiter_bytes(), content_length, options + ) for request in httpx_mock.get_requests(): assert "vaasrequestid=foobar" in request.headers["tracestate"] @@ -351,21 +362,22 @@ async def test_for_stream_set_request_id_send_trace_state(self, vaas, httpx_mock assert verdict.verdict == "Clean" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_stream_bad_request_raise_vaas_client_error(self, vaas, httpx_mock): + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_stream_bad_request_raise_vaas_client_error( + self, vaas, httpx_mock + ): httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=400, - json = { - "detail": "Mocked client-side error", - "type": "VaasClientException" - } + json={"detail": "Mocked client-side error", "type": "VaasClientException"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] with pytest.raises(VaasClientError) as exception_info: await vaas.for_stream(response.aiter_bytes(), content_length) @@ -375,21 +387,22 @@ async def test_for_stream_bad_request_raise_vaas_client_error(self, vaas, httpx_ assert problem_details.type == "VaasClientException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_stream_server_error_raise_vaas_server_error(self, vaas, httpx_mock): + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_stream_server_error_raise_vaas_server_error( + self, vaas, httpx_mock + ): httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=500, - json = { - "detail": "Mocked server-side error", - "type": "VaasServerException" - } + json={"detail": "Mocked server-side error", "type": "VaasServerException"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] with pytest.raises(VaasServerError) as exception_info: await vaas.for_stream(response.aiter_bytes(), content_length) @@ -399,32 +412,42 @@ async def test_for_stream_server_error_raise_vaas_server_error(self, vaas, httpx assert problem_details.type == "VaasServerException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_stream_authentication_error_raise_vaas_authentication_error(self, vaas, httpx_mock): - vaas.authenticator.get_token = AsyncMock(side_effect=VaasAuthenticationError("Mocked auth error")) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_stream_authentication_error_raise_vaas_authentication_error( + self, vaas, httpx_mock + ): + vaas.authenticator.get_token = AsyncMock( + side_effect=VaasAuthenticationError("Mocked auth error") + ) async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] with pytest.raises(VaasAuthenticationError): await vaas.for_stream(response.aiter_bytes(), content_length) @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_stream_unauthorized_raise_vaas_authentication_error(self, vaas, httpx_mock): + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_stream_unauthorized_raise_vaas_authentication_error( + self, vaas, httpx_mock + ): httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=401, - json = { + json={ "detail": "Authentication error", - "type": "VaasAuthenticationException" - } + "type": "VaasAuthenticationException", + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] with pytest.raises(VaasAuthenticationError) as exception_info: await vaas.for_stream(response.aiter_bytes(), content_length) @@ -434,45 +457,53 @@ async def test_for_stream_unauthorized_raise_vaas_authentication_error(self, vaa assert problem_details.type == "VaasAuthenticationException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_stream_cancel_request_raise_cancel_error(self, vaas, httpx_mock): vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") httpx_mock.add_exception( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", - exception=asyncio.CancelledError() + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + exception=asyncio.CancelledError(), ) async with httpx.AsyncClient() as client: - response = await client.get("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt") + response = await client.get(CLEAN_URL) content_length = response.headers["Content-Length"] with pytest.raises(asyncio.CancelledError): await vaas.for_stream(response.aiter_bytes(), content_length) - @pytest.mark.asyncio() @pytest.mark.parametrize( - "url, expected_verdict", + "url, expected_verdict, is_encrypted", [ - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/eicar.com.txt", "Malicious"), - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt", "Clean"), - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/PotentiallyUnwanted.exe", "Pup") + (EICAR_URL, "Malicious", False), + (CLEAN_URL, "Clean", False), + (PUP_URL, "Pup", False), + (PASSWORD_ZIP_URL, "Clean", True), + (WITH_AND_WITHOUT_PASSWORD_ZIP_URL, "Malicious", True), ], - ids=["Malware", "Clean", "Pup"] + ids=["Malware", "Clean", "Pup", "Encrypted", "EncryptedWithMalwareInside"], ) - async def test_for_file_returns_verdict(self, vaas, url, expected_verdict): - async with httpx.AsyncClient() as client: - filename = os.path.join("/tmp", os.path.basename(url)) - response = await client.get(url) - response.raise_for_status() - with open(filename, mode="wb") as file: - file.write(response.content) - verdict = await vaas.for_file(filename) + async def test_for_file_returns_verdict( + self, vaas, url, expected_verdict, is_encrypted + ): + async with httpx.AsyncClient() as client: + filename = os.path.join("/tmp", os.path.basename(url)) + response = await client.get(url) + response.raise_for_status() + with open(filename, mode="wb") as file: + file.write(response.content) + verdict = await vaas.for_file(filename) - assert verdict.verdict == expected_verdict + assert verdict.verdict == expected_verdict + assert verdict.isEncrypted == is_encrypted @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) @pytest.mark.parametrize( ("use_cache", "use_hash_lookup", "request_count"), [ @@ -481,10 +512,12 @@ async def test_for_file_returns_verdict(self, vaas, url, expected_verdict): (True, False, 3), (True, True, 3), ], - ids=["false_for_all", "only_hash_lookup", "only_cache", "true_for_all"] + ids=["false_for_all", "only_hash_lookup", "only_cache", "true_for_all"], ) - async def test_for_file_send_options(self, vaas, use_cache, use_hash_lookup, request_count, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + async def test_for_file_send_options( + self, vaas, use_cache, use_hash_lookup, request_count, httpx_mock + ): + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_response( @@ -496,9 +529,9 @@ async def test_for_file_send_options(self, vaas, use_cache, use_hash_lookup, req "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None + "mimeType": None, }, - is_optional=True + is_optional=True, ) httpx_mock.add_response( method="GET", @@ -509,16 +542,14 @@ async def test_for_file_send_options(self, vaas, use_cache, use_hash_lookup, req "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/files?useHashLookup={str(use_hash_lookup).lower()}", status_code=200, - json={ - "sha256": CLEAN_SHA256 - } + json={"sha256": CLEAN_SHA256}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -535,9 +566,11 @@ async def test_for_file_send_options(self, vaas, use_cache, use_hash_lookup, req assert len(httpx_mock.get_requests()) == request_count @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_file_send_user_agent(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_response( @@ -549,17 +582,15 @@ async def test_for_file_send_user_agent(self, vaas, httpx_mock): "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None + "mimeType": None, }, - is_reusable=True + is_reusable=True, ) httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/files?useHashLookup=true", status_code=200, - json={ - "sha256": CLEAN_SHA256 - } + json={"sha256": CLEAN_SHA256}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -578,9 +609,11 @@ async def test_for_file_send_user_agent(self, vaas, httpx_mock): assert len(httpx_mock.get_requests()) == 3 @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_file_set_request_id_send_trace_state(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_response( @@ -592,17 +625,15 @@ async def test_for_file_set_request_id_send_trace_state(self, vaas, httpx_mock): "verdict": "Clean", "detection": None, "fileType": None, - "mimeType": None + "mimeType": None, }, - is_reusable=True + is_reusable=True, ) httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/files?useHashLookup=true", status_code=200, - json={ - "sha256": CLEAN_SHA256 - } + json={"sha256": CLEAN_SHA256}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -621,29 +652,25 @@ async def test_for_file_set_request_id_send_trace_state(self, vaas, httpx_mock): assert verdict.verdict == "Clean" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_file_bad_request_raise_vaas_client_error(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_response( method="GET", url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache=true&useHashLookup=true", status_code=400, - json = { - "detail": "Mocked client-side error", - "type": "VaasClientException" - } + json={"detail": "Mocked client-side error", "type": "VaasClientException"}, ) httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=400, - json = { - "detail": "Mocked client-side error", - "type": "VaasClientException" - } + json={"detail": "Mocked client-side error", "type": "VaasClientException"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -662,29 +689,27 @@ async def test_for_file_bad_request_raise_vaas_client_error(self, vaas, httpx_mo assert problem_details.type == "VaasClientException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_file_server_error_raise_vaas_server_error(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_file_server_error_raise_vaas_server_error( + self, vaas, httpx_mock + ): + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_response( method="GET", url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache=true&useHashLookup=true", status_code=500, - json = { - "detail": "Mocked server-side error", - "type": "VaasServerException" - } + json={"detail": "Mocked server-side error", "type": "VaasServerException"}, ) httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=500, - json = { - "detail": "Mocked server-side error", - "type": "VaasServerException" - } + json={"detail": "Mocked server-side error", "type": "VaasServerException"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -702,11 +727,15 @@ async def test_for_file_server_error_raise_vaas_server_error(self, vaas, httpx_m assert problem_details.type == "VaasServerException" @pytest.mark.asyncio() - async def test_for_file_authentication_error_raise_vaas_authentication_error(self, vaas): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + async def test_for_file_authentication_error_raise_vaas_authentication_error( + self, vaas + ): + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) - vaas.authenticator.get_token = AsyncMock(side_effect=VaasAuthenticationError("Mocked auth error")) + vaas.authenticator.get_token = AsyncMock( + side_effect=VaasAuthenticationError("Mocked auth error") + ) async with httpx.AsyncClient() as client: response = await client.get(url) @@ -718,9 +747,13 @@ async def test_for_file_authentication_error_raise_vaas_authentication_error(sel await vaas.for_sha256(CLEAN_SHA256) @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_file_unauthorized_raise_vaas_authentication_error(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_file_unauthorized_raise_vaas_authentication_error( + self, vaas, httpx_mock + ): + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_response( @@ -729,18 +762,18 @@ async def test_for_file_unauthorized_raise_vaas_authentication_error(self, vaas, status_code=401, json={ "detail": "Authentication error", - "type": "VaasAuthenticationException" - } + "type": "VaasAuthenticationException", + }, ) httpx_mock.add_response( method="POST", - url= f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", + url=f"{VAAS_URL}/files?useHashLookup={str(True).lower()}", status_code=401, - json = { + json={ "detail": "Authentication error", - "type": "VaasAuthenticationException" - } + "type": "VaasAuthenticationException", + }, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -759,15 +792,17 @@ async def test_for_file_unauthorized_raise_vaas_authentication_error(self, vaas, assert problem_details.type == "VaasAuthenticationException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_file_cancel_request_raise_cancel_error(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL filename = os.path.join("/tmp", os.path.basename(url)) httpx_mock.add_exception( method="GET", url=f"{VAAS_URL}/files/{CLEAN_SHA256}/report?useCache=true&useHashLookup=true", - exception=asyncio.CancelledError() + exception=asyncio.CancelledError(), ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -783,31 +818,35 @@ async def test_for_file_cancel_request_raise_cancel_error(self, vaas, httpx_mock @pytest.mark.asyncio() @pytest.mark.parametrize( - "url, expected_verdict", + "url, expected_verdict, is_encrypted", [ - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/eicar.com.txt", "Malicious"), - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt", "Clean"), - ("https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/PotentiallyUnwanted.exe", "Pup") + (EICAR_URL, "Malicious", False), + (CLEAN_URL, "Clean", False), + (PUP_URL, "Pup", False), + (PASSWORD_ZIP_URL, "Clean", True), + (WITH_AND_WITHOUT_PASSWORD_ZIP_URL, "Malicious", True), ], - ids=["Malware", "Clean", "Pup"] + ids=["Malware", "Clean", "Pup", "Encrypted", "EncryptedWithMalwareInside"], ) - async def test_for_url_returns_verdict(self, vaas, url, expected_verdict): + async def test_for_url_returns_verdict( + self, vaas, url, expected_verdict, is_encrypted + ): verdict = await vaas.for_url(url) assert verdict.verdict == expected_verdict + assert verdict.isEncrypted == is_encrypted @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) @pytest.mark.parametrize( "use_hash_lookup", - [ - False, - True - ], - ids=["hash_lookup_enabled", "hash_lookup_disabled"] + [False, True], + ids=["hash_lookup_enabled", "hash_lookup_disabled"], ) async def test_for_url_send_options(self, vaas, use_hash_lookup, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL httpx_mock.add_response( method="GET", @@ -819,16 +858,14 @@ async def test_for_url_send_options(self, vaas, use_hash_lookup, httpx_mock): "url": url, "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - } + json={"id": "foobar"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -840,9 +877,11 @@ async def test_for_url_send_options(self, vaas, use_hash_lookup, httpx_mock): assert len(httpx_mock.get_requests()) == 2 @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_url_send_user_agent(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL httpx_mock.add_response( method="GET", @@ -854,16 +893,14 @@ async def test_for_url_send_user_agent(self, vaas, httpx_mock): "url": url, "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - } + json={"id": "foobar"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -871,14 +908,15 @@ async def test_for_url_send_user_agent(self, vaas, httpx_mock): for request in httpx_mock.get_requests(): assert "Python" in request.headers["User-Agent"] - assert verdict.verdict == "Clean" assert len(httpx_mock.get_requests()) == 2 @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) async def test_for_url_set_request_id_send_trace_state(self, vaas, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + url = CLEAN_URL httpx_mock.add_response( method="GET", @@ -890,16 +928,14 @@ async def test_for_url_set_request_id_send_trace_state(self, vaas, httpx_mock): "url": url, "detection": None, "fileType": None, - "mimeType": None - } + "mimeType": None, + }, ) httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - } + json={"id": "foobar"}, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -913,10 +949,14 @@ async def test_for_url_set_request_id_send_trace_state(self, vaas, httpx_mock): assert verdict.verdict == "Clean" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) @pytest.mark.parametrize("post_fails", [True, False], ids=["post_400", "get_400"]) - async def test_for_url_bad_request_raise_vaas_client_error(self, vaas, httpx_mock, post_fails): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + async def test_for_url_bad_request_raise_vaas_client_error( + self, vaas, httpx_mock, post_fails + ): + url = CLEAN_URL if post_fails: httpx_mock.add_response( @@ -925,19 +965,17 @@ async def test_for_url_bad_request_raise_vaas_client_error(self, vaas, httpx_moc status_code=400, json={ "detail": "Mocked client-side error (POST)", - "type": "VaasClientException" + "type": "VaasClientException", }, - is_optional=True + is_optional=True, ) else: httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - }, - is_optional = True + json={"id": "foobar"}, + is_optional=True, ) httpx_mock.add_response( @@ -946,9 +984,9 @@ async def test_for_url_bad_request_raise_vaas_client_error(self, vaas, httpx_moc status_code=400, json={ "detail": "Mocked client-side error (GET)", - "type": "VaasClientException" + "type": "VaasClientException", }, - is_optional=True + is_optional=True, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -961,10 +999,14 @@ async def test_for_url_bad_request_raise_vaas_client_error(self, vaas, httpx_moc assert problem_details.type == "VaasClientException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) @pytest.mark.parametrize("post_fails", [True, False], ids=["post_500", "get_500"]) - async def test_for_url_server_error_raise_vaas_server_error(self, vaas, httpx_mock, post_fails): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + async def test_for_url_server_error_raise_vaas_server_error( + self, vaas, httpx_mock, post_fails + ): + url = CLEAN_URL if post_fails: httpx_mock.add_response( @@ -973,19 +1015,17 @@ async def test_for_url_server_error_raise_vaas_server_error(self, vaas, httpx_mo status_code=500, json={ "detail": "Mocked server-side error (POST)", - "type": "VaasServerException" + "type": "VaasServerException", }, - is_optional=True + is_optional=True, ) else: httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - }, - is_optional = True + json={"id": "foobar"}, + is_optional=True, ) httpx_mock.add_response( @@ -994,9 +1034,9 @@ async def test_for_url_server_error_raise_vaas_server_error(self, vaas, httpx_mo status_code=500, json={ "detail": "Mocked server-side error (GET)", - "type": "VaasServerException" + "type": "VaasServerException", }, - is_optional=True + is_optional=True, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -1009,18 +1049,28 @@ async def test_for_url_server_error_raise_vaas_server_error(self, vaas, httpx_mo assert problem_details.type == "VaasServerException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - async def test_for_url_authentication_error_raise_vaas_authentication_error(self, vaas): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" - vaas.authenticator.get_token = AsyncMock(side_effect=VaasAuthenticationError("Mocked auth error")) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + async def test_for_url_authentication_error_raise_vaas_authentication_error( + self, vaas + ): + url = CLEAN_URL + vaas.authenticator.get_token = AsyncMock( + side_effect=VaasAuthenticationError("Mocked auth error") + ) with pytest.raises(VaasAuthenticationError): await vaas.for_url(url) @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) @pytest.mark.parametrize("post_fails", [True, False], ids=["post_401", "get_401"]) - async def test_for_url_unauthorized_raise_vaas_authentication_error(self, vaas, post_fails, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + async def test_for_url_unauthorized_raise_vaas_authentication_error( + self, vaas, post_fails, httpx_mock + ): + url = CLEAN_URL if post_fails: httpx_mock.add_response( @@ -1029,19 +1079,17 @@ async def test_for_url_unauthorized_raise_vaas_authentication_error(self, vaas, status_code=401, json={ "detail": "Authentication error", - "type": "VaasAuthenticationException" + "type": "VaasAuthenticationException", }, - is_optional=True + is_optional=True, ) else: httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - }, - is_optional = True + json={"id": "foobar"}, + is_optional=True, ) httpx_mock.add_response( @@ -1050,9 +1098,9 @@ async def test_for_url_unauthorized_raise_vaas_authentication_error(self, vaas, status_code=401, json={ "detail": "Authentication error", - "type": "VaasAuthenticationException" + "type": "VaasAuthenticationException", }, - is_optional=True + is_optional=True, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token") @@ -1065,34 +1113,38 @@ async def test_for_url_unauthorized_raise_vaas_authentication_error(self, vaas, assert problem_details.type == "VaasAuthenticationException" @pytest.mark.asyncio() - @pytest.mark.httpx_mock(should_mock=lambda request: "gdatasecurity.de" in request.url.host) - @pytest.mark.parametrize("post_fails", [True, False], ids=["post_cancel", "get_cancel"]) - async def test_for_url_cancel_request_raise_cancel_error(self, vaas, post_fails, httpx_mock): - url = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt" + @pytest.mark.httpx_mock( + should_mock=lambda request: "gdatasecurity.de" in request.url.host + ) + @pytest.mark.parametrize( + "post_fails", [True, False], ids=["post_cancel", "get_cancel"] + ) + async def test_for_url_cancel_request_raise_cancel_error( + self, vaas, post_fails, httpx_mock + ): + url = CLEAN_URL if post_fails: httpx_mock.add_exception( method="POST", url=f"{VAAS_URL}/urls", exception=asyncio.CancelledError(), - is_optional = True + is_optional=True, ) else: httpx_mock.add_response( method="POST", url=f"{VAAS_URL}/urls", status_code=200, - json={ - "id": "foobar" - }, - is_optional = True + json={"id": "foobar"}, + is_optional=True, ) httpx_mock.add_exception( method="GET", url=f"{VAAS_URL}/urls/foobar/report", exception=asyncio.CancelledError(), - is_optional = True + is_optional=True, ) vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")