✨ source-file: prevent local file usage on cloud deployment mode (#30984)

Co-authored-by: pedroslopez <pedroslopez@users.noreply.github.com>

Author: pedroslopez

airbyte-integrations/connectors/source-file/acceptance-test-config.yml

@@ -6,12 +6,17 @@ acceptance_tests:
  spec:
  tests:
  - spec_path: "source_file/spec.json"
+ - spec_path: "integration_tests/cloud_spec.json"
+ deployment_mode: "cloud"
  connection:
  tests:
  - config_path: "integration_tests/config.json"
  status: "succeed"
  - config_path: "integration_tests/invalid_config.json"
  status: "failed"
+ - config_path: "integration_tests/local_config.json"
+ deployment_mode: "cloud"
+ status: "failed"
  discovery:
  tests:
  - config_path: "integration_tests/config.json"

airbyte-integrations/connectors/source-file/integration_tests/cloud_spec.json

@@ -0,0 +1,216 @@
+{
+ "documentationUrl": "https://docs.airbyte.com/integrations/sources/file",
+ "connectionSpecification": {
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "title": "File Source Spec",
+ "type": "object",
+ "additionalProperties": true,
+ "required": ["dataset_name", "format", "url", "provider"],
+ "properties": {
+ "dataset_name": {
+ "type": "string",
+ "title": "Dataset Name",
+ "description": "The Name of the final table to replicate this file into (should include letters, numbers dash and underscores only)."
+ },
+ "format": {
+ "type": "string",
+ "enum": [
+ "csv",
+ "json",
+ "jsonl",
+ "excel",
+ "excel_binary",
+ "feather",
+ "parquet",
+ "yaml"
+ ],
+ "default": "csv",
+ "title": "File Format",
+ "description": "The Format of the file which should be replicated (Warning: some formats may be experimental, please refer to the docs)."
+ },
+ "reader_options": {
+ "type": "string",
+ "title": "Reader Options",
+ "description": "This should be a string in JSON format. It depends on the chosen file format to provide additional options and tune its behavior.",
+ "examples": [
+ "{}",
+ "{\"sep\": \" \"}",
+ "{\"sep\": \"\t\", \"header\": 0, \"names\": [\"column1\", \"column2\"] }"
+ ]
+ },
+ "url": {
+ "type": "string",
+ "title": "URL",
+ "description": "The URL path to access the file which should be replicated.",
+ "examples": [
+ "https://storage.googleapis.com/covid19-open-data/v2/latest/epidemiology.csv",
+ "gs://my-google-bucket/data.csv",
+ "s3://gdelt-open-data/events/20190914.export.csv"
+ ]
+ },
+ "provider": {
+ "type": "object",
+ "title": "Storage Provider",
+ "description": "The storage Provider or Location of the file(s) which should be replicated.",
+ "default": "Public Web",
+ "oneOf": [
+ {
+ "title": "HTTPS: Public Web",
+ "required": ["storage"],
+ "properties": {
+ "storage": { "type": "string", "const": "HTTPS" },
+ "user_agent": {
+ "type": "boolean",
+ "title": "User-Agent",
+ "default": false,
+ "description": "Add User-Agent to request"
+ }
+ }
+ },
+ {
+ "title": "GCS: Google Cloud Storage",
+ "required": ["storage"],
+ "properties": {
+ "storage": {
+ "type": "string",
+ "title": "Storage",
+ "const": "GCS"
+ },
+ "service_account_json": {
+ "type": "string",
+ "title": "Service Account JSON",
+ "airbyte_secret": true,
+ "description": "In order to access private Buckets stored on Google Cloud, this connector would need a service account json credentials with the proper permissions as described <a href=\"https://cloud.google.com/iam/docs/service-accounts\" target=\"_blank\">here</a>. Please generate the credentials.json file and copy/paste its content to this field (expecting JSON formats). If accessing publicly available data, this field is not necessary."
+ }
+ }
+ },
+ {
+ "title": "S3: Amazon Web Services",
+ "required": ["storage"],
+ "properties": {
+ "storage": {
+ "type": "string",
+ "title": "Storage",
+ "const": "S3"
+ },
+ "aws_access_key_id": {
+ "type": "string",
+ "title": "AWS Access Key ID",
+ "description": "In order to access private Buckets stored on AWS S3, this connector would need credentials with the proper permissions. If accessing publicly available data, this field is not necessary."
+ },
+ "aws_secret_access_key": {
+ "type": "string",
+ "title": "AWS Secret Access Key",
+ "description": "In order to access private Buckets stored on AWS S3, this connector would need credentials with the proper permissions. If accessing publicly available data, this field is not necessary.",
+ "airbyte_secret": true
+ }
+ }
+ },
+ {
+ "title": "AzBlob: Azure Blob Storage",
+ "required": ["storage", "storage_account"],
+ "properties": {
+ "storage": {
+ "type": "string",
+ "title": "Storage",
+ "const": "AzBlob"
+ },
+ "storage_account": {
+ "type": "string",
+ "title": "Storage Account",
+ "description": "The globally unique name of the storage account that the desired blob sits within. See <a href=\"https://docs.microsoft.com/en-us/azure/storage/common/storage-account-overview\" target=\"_blank\">here</a> for more details."
+ },
+ "sas_token": {
+ "type": "string",
+ "title": "SAS Token",
+ "description": "To access Azure Blob Storage, this connector would need credentials with the proper permissions. One option is a SAS (Shared Access Signature) token. If accessing publicly available data, this field is not necessary.",
+ "airbyte_secret": true
+ },
+ "shared_key": {
+ "type": "string",
+ "title": "Shared Key",
+ "description": "To access Azure Blob Storage, this connector would need credentials with the proper permissions. One option is a storage account shared key (aka account key or access key). If accessing publicly available data, this field is not necessary.",
+ "airbyte_secret": true
+ }
+ }
+ },
+ {
+ "title": "SSH: Secure Shell",
+ "required": ["storage", "user", "host"],
+ "properties": {
+ "storage": {
+ "type": "string",
+ "title": "Storage",
+ "const": "SSH"
+ },
+ "user": { "type": "string", "title": "User", "description": "" },
+ "password": {
+ "type": "string",
+ "title": "Password",
+ "description": "",
+ "airbyte_secret": true
+ },
+ "host": { "type": "string", "title": "Host", "description": "" },
+ "port": {
+ "type": "string",
+ "title": "Port",
+ "default": "22",
+ "description": ""
+ }
+ }
+ },
+ {
+ "title": "SCP: Secure copy protocol",
+ "required": ["storage", "user", "host"],
+ "properties": {
+ "storage": {
+ "type": "string",
+ "title": "Storage",
+ "const": "SCP"
+ },
+ "user": { "type": "string", "title": "User", "description": "" },
+ "password": {
+ "type": "string",
+ "title": "Password",
+ "description": "",
+ "airbyte_secret": true
+ },
+ "host": { "type": "string", "title": "Host", "description": "" },
+ "port": {
+ "type": "string",
+ "title": "Port",
+ "default": "22",
+ "description": ""
+ }
+ }
+ },
+ {
+ "title": "SFTP: Secure File Transfer Protocol",
+ "required": ["storage", "user", "host"],
+ "properties": {
+ "storage": {
+ "type": "string",
+ "title": "Storage",
+ "const": "SFTP"
+ },
+ "user": { "type": "string", "title": "User", "description": "" },
+ "password": {
+ "type": "string",
+ "title": "Password",
+ "description": "",
+ "airbyte_secret": true
+ },
+ "host": { "type": "string", "title": "Host", "description": "" },
+ "port": {
+ "type": "string",
+ "title": "Port",
+ "default": "22",
+ "description": ""
+ }
+ }
+ }
+ ]
+ }
+ }
+ }
+}

airbyte-integrations/connectors/source-file/integration_tests/local_config.json

@@ -0,0 +1,9 @@
+{
+ "dataset_name": "test",
+ "format": "csv",
+ "reader_options": "{\"bla\": \",\", \"nrows\": 20}",
+ "url": "file:///tmp/fake_file.csv",
+ "provider": {
+ "storage": "local"
+ }
+}

airbyte-integrations/connectors/source-file/metadata.yaml

@@ -7,16 +7,14 @@ data:
  connectorSubtype: file
  connectorType: source
  definitionId: 778daa7c-feaf-4db6-96f3-70fd645acc77
- dockerImageTag: 0.3.13
+ dockerImageTag: 0.3.14
  dockerRepository: airbyte/source-file
  githubIssueLabel: source-file
  icon: file.svg
  license: MIT
  name: File (CSV, JSON, Excel, Feather, Parquet)
  registries:
  cloud:
- dockerRepository: airbyte/source-file-secure
- dockerImageTag: 0.3.13 # Dont forget to publish source-file-secure as well when updating this.
  enabled: true
  oss:
  enabled: true

airbyte-integrations/connectors/source-file/setup.py

@@ -6,7 +6,7 @@
 from setuptools import find_packages, setup
 
 MAIN_REQUIREMENTS = [
- "airbyte-cdk~=0.2",
+ "airbyte-cdk~=0.51.25",
  "gcsfs==2022.7.1",
  "genson==1.2.2",
  "google-cloud-storage==2.5.0",

airbyte-integrations/connectors/source-file/source_file/client.py

@@ -24,7 +24,7 @@
 import smart_open.ssh
 from airbyte_cdk.entrypoint import logger
 from airbyte_cdk.models import AirbyteStream, FailureType, SyncMode
-from airbyte_cdk.utils import AirbyteTracedException
+from airbyte_cdk.utils import AirbyteTracedException, is_cloud_environment
 from azure.storage.blob import BlobServiceClient
 from genson import SchemaBuilder
 from google.cloud.storage import Client as GCSClient
@@ -36,7 +36,7 @@
 from urllib3.exceptions import ProtocolError
 from yaml import safe_load
 
-from .utils import backoff_handler
+from .utils import LOCAL_STORAGE_NAME, backoff_handler
 
 SSH_TIMEOUT = 60
 
@@ -252,7 +252,6 @@ class Client:
  """Class that manages reading and parsing data from streams"""
 
  CSV_CHUNK_SIZE = 10_000
- reader_class = URLFile
  binary_formats = {"excel", "excel_binary", "feather", "parquet", "orc", "pickle"}
 
  def __init__(self, dataset_name: str, url: str, provider: dict, format: str = None, reader_options: dict = None):
@@ -264,6 +263,13 @@ def __init__(self, dataset_name: str, url: str, provider: dict, format: str = No
  self.binary_source = self._reader_format in self.binary_formats
  self.encoding = self._reader_options.get("encoding")
 
+ @property
+ def reader_class(self):
+ if is_cloud_environment():
+ return URLFileSecure
+
+ return URLFile
+
  @property
  def stream_name(self) -> str:
  if self._dataset_name:
@@ -497,3 +503,15 @@ def openpyxl_chunk_reader(self, file, **kwargs):
  df = pd.DataFrame(data=(next(data) for _ in range(start, min(start + step, end))), columns=cols)
  yield df
  start += step
+
+
+class URLFileSecure(URLFile):
+ """Updating of default logic:
+ This connector shouldn't work with local files.
+ """
+
+ def __init__(self, url: str, provider: dict, binary=None, encoding=None):
+ storage_name = provider["storage"].lower()
+ if url.startswith("file://") or storage_name == LOCAL_STORAGE_NAME:
+ raise RuntimeError("the local file storage is not supported by this connector.")
+ super().__init__(url, provider, binary, encoding)

airbyte-integrations/connectors/source-file/source_file/source.py

@@ -17,15 +17,16 @@
  AirbyteMessage,
  AirbyteRecordMessage,
  ConfiguredAirbyteCatalog,
+ ConnectorSpecification,
  FailureType,
  Status,
  Type,
 )
 from airbyte_cdk.sources import Source
-from airbyte_cdk.utils import AirbyteTracedException
+from airbyte_cdk.utils import AirbyteTracedException, is_cloud_environment
 
 from .client import Client
-from .utils import dropbox_force_download
+from .utils import LOCAL_STORAGE_NAME, dropbox_force_download
 
 
 class SourceFile(Source):
@@ -105,6 +106,19 @@ def _validate_and_transform(config: Mapping[str, Any]):
  raise AirbyteTracedException(message=message, internal_message=message, failure_type=FailureType.config_error)
  return config
 
+ def spec(self, logger: AirbyteLogger) -> ConnectorSpecification:
+ """Returns the json schema for the spec"""
+ spec = super().spec(logger)
+
+ # override cloud spec to remove local file support
+ if is_cloud_environment():
+ for i in range(len(spec.connectionSpecification["properties"]["provider"]["oneOf"])):
+ provider = spec.connectionSpecification["properties"]["provider"]["oneOf"][i]
+ if provider["properties"]["storage"]["const"] == LOCAL_STORAGE_NAME:
+ spec.connectionSpecification["properties"]["provider"]["oneOf"].pop(i)
+
+ return spec
+
  def check(self, logger, config: Mapping) -> AirbyteConnectionStatus:
  """
  Check involves verifying that the specified file is reachable with

airbyte-integrations/connectors/source-file/source_file/utils.py

@@ -8,6 +8,8 @@
 # default logger
 logger = logging.getLogger("airbyte")
 
+LOCAL_STORAGE_NAME = "local"
+
 
 def dropbox_force_download(url):
  """