fix copilot suggestions

Signed-off-by: hassaansaleem28 <iamhassaans@gmail.com>

Author: hassaansaleem28

ingestion/src/metadata/ingestion/source/database/oracle/connection.py

@@ -13,6 +13,7 @@
 Source connection handler
 """
 import base64
+import binascii
 import io
 import os
 import shutil
@@ -21,7 +22,7 @@
 import weakref
 import zipfile
 from copy import deepcopy
-from typing import Any, Optional
+from typing import Optional
 from urllib.parse import quote_plus
 
 import oracledb
@@ -106,7 +107,7 @@ def _is_autonomous_connection(self) -> bool:
     @staticmethod
     def _get_autonomous_connection_config(
         connection_type: OracleAutonomousConnection,
-    ) -> Any:
+    ) -> OracleAutonomousConnection:
         return connection_type
 
     @staticmethod
@@ -144,8 +145,10 @@ def _safe_extract_wallet_archive(zip_ref: zipfile.ZipFile, target_dir: str) -> N
 
     def _extract_wallet_content(self, wallet_content: SecretStr) -> str:
         try:
-            decoded_wallet = base64.b64decode(wallet_content.get_secret_value())
-        except (ValueError, TypeError) as exc:
+            decoded_wallet = base64.b64decode(
+                wallet_content.get_secret_value(), validate=True
+            )
+        except (binascii.Error, TypeError) as exc:
             raise ValueError(
                 "Invalid walletContent. Expected a base64-encoded wallet zip."
             ) from exc

ingestion/tests/unit/test_source_connection.py

@@ -1373,6 +1373,29 @@ def test_oracle_autonomous_wallet_content_zip_slip_rejected(
         assert "unsafe file paths" in str(error.exception)
         assert oracle_connection._wallet_temp_dir is None
 
+    @patch(
+        "metadata.ingestion.source.database.oracle.connection.create_generic_db_connection"
+    )
+    def test_oracle_autonomous_wallet_content_invalid_base64_rejected(
+        self, mock_create_generic_db_connection
+    ):
+        connection = OracleConnectionConfig(
+            username="admin",
+            password="password",
+            oracleConnectionType=OracleAutonomousConnection(
+                tnsAlias="myadb_high",
+                walletContent="Zm9v$",
+            ),
+        )
+        oracle_connection = OracleConnection(connection)
+        mock_create_generic_db_connection.return_value = "dummy_engine"
+
+        with self.assertRaises(ValueError) as error:
+            oracle_connection._get_client()
+
+        assert "base64-encoded wallet zip" in str(error.exception)
+        assert oracle_connection._wallet_temp_dir is None
+
     def test_exasol_url(self):
         from metadata.ingestion.source.database.exasol.connection import (
             get_connection_url,