diff options
Diffstat (limited to 'activestorage/lib/active_storage/service')
6 files changed, 683 insertions, 0 deletions
diff --git a/activestorage/lib/active_storage/service/azure_storage_service.rb b/activestorage/lib/active_storage/service/azure_storage_service.rb new file mode 100644 index 0000000000..8de3889cb5 --- /dev/null +++ b/activestorage/lib/active_storage/service/azure_storage_service.rb @@ -0,0 +1,167 @@ +# frozen_string_literal: true + +require "active_support/core_ext/numeric/bytes" +require "azure/storage" +require "azure/storage/core/auth/shared_access_signature" + +module ActiveStorage + # Wraps the Microsoft Azure Storage Blob Service as an Active Storage service. + # See ActiveStorage::Service for the generic API documentation that applies to all services. + class Service::AzureStorageService < Service + attr_reader :client, :blobs, :container, :signer + + def initialize(storage_account_name:, storage_access_key:, container:) + @client = Azure::Storage::Client.create(storage_account_name: storage_account_name, storage_access_key: storage_access_key) + @signer = Azure::Storage::Core::Auth::SharedAccessSignature.new(storage_account_name, storage_access_key) + @blobs = client.blob_client + @container = container + end + + def upload(key, io, checksum: nil) + instrument :upload, key: key, checksum: checksum do + handle_errors do + blobs.create_block_blob(container, key, IO.try_convert(io) || io, content_md5: checksum) + end + end + end + + def download(key, &block) + if block_given? + instrument :streaming_download, key: key do + stream(key, &block) + end + else + instrument :download, key: key do + handle_errors do + _, io = blobs.get_blob(container, key) + io.force_encoding(Encoding::BINARY) + end + end + end + end + + def download_chunk(key, range) + instrument :download_chunk, key: key, range: range do + handle_errors do + _, io = blobs.get_blob(container, key, start_range: range.begin, end_range: range.exclude_end? ? range.end - 1 : range.end) + io.force_encoding(Encoding::BINARY) + end + end + end + + def delete(key) + instrument :delete, key: key do + begin + blobs.delete_blob(container, key) + rescue Azure::Core::Http::HTTPError => e + raise unless e.type == "BlobNotFound" + # Ignore files already deleted + end + end + end + + def delete_prefixed(prefix) + instrument :delete_prefixed, prefix: prefix do + marker = nil + + loop do + results = blobs.list_blobs(container, prefix: prefix, marker: marker) + + results.each do |blob| + blobs.delete_blob(container, blob.name) + end + + break unless marker = results.continuation_token.presence + end + end + end + + def exist?(key) + instrument :exist, key: key do |payload| + answer = blob_for(key).present? + payload[:exist] = answer + answer + end + end + + def url(key, expires_in:, filename:, disposition:, content_type:) + instrument :url, key: key do |payload| + generated_url = signer.signed_uri( + uri_for(key), false, + service: "b", + permissions: "r", + expiry: format_expiry(expires_in), + content_disposition: content_disposition_with(type: disposition, filename: filename), + content_type: content_type + ).to_s + + payload[:url] = generated_url + + generated_url + end + end + + def url_for_direct_upload(key, expires_in:, content_type:, content_length:, checksum:) + instrument :url, key: key do |payload| + generated_url = signer.signed_uri( + uri_for(key), false, + service: "b", + permissions: "rw", + expiry: format_expiry(expires_in) + ).to_s + + payload[:url] = generated_url + + generated_url + end + end + + def headers_for_direct_upload(key, content_type:, checksum:, **) + { "Content-Type" => content_type, "Content-MD5" => checksum, "x-ms-blob-type" => "BlockBlob" } + end + + private + def uri_for(key) + blobs.generate_uri("#{container}/#{key}") + end + + def blob_for(key) + blobs.get_blob_properties(container, key) + rescue Azure::Core::Http::HTTPError + false + end + + def format_expiry(expires_in) + expires_in ? Time.now.utc.advance(seconds: expires_in).iso8601 : nil + end + + # Reads the object for the given key in chunks, yielding each to the block. + def stream(key) + blob = blob_for(key) + + chunk_size = 5.megabytes + offset = 0 + + raise ActiveStorage::FileNotFoundError unless blob.present? + + while offset < blob.properties[:content_length] + _, chunk = blobs.get_blob(container, key, start_range: offset, end_range: offset + chunk_size - 1) + yield chunk.force_encoding(Encoding::BINARY) + offset += chunk_size + end + end + + def handle_errors + yield + rescue Azure::Core::Http::HTTPError => e + case e.type + when "BlobNotFound" + raise ActiveStorage::FileNotFoundError + when "Md5Mismatch" + raise ActiveStorage::IntegrityError + else + raise + end + end + end +end diff --git a/activestorage/lib/active_storage/service/configurator.rb b/activestorage/lib/active_storage/service/configurator.rb new file mode 100644 index 0000000000..fa80c66c3b --- /dev/null +++ b/activestorage/lib/active_storage/service/configurator.rb @@ -0,0 +1,34 @@ +# frozen_string_literal: true + +module ActiveStorage + class Service::Configurator #:nodoc: + attr_reader :configurations + + def self.build(service_name, configurations) + new(configurations).build(service_name) + end + + def initialize(configurations) + @configurations = configurations.deep_symbolize_keys + end + + def build(service_name) + config = config_for(service_name.to_sym) + resolve(config.fetch(:service)).build(**config, configurator: self) + end + + private + def config_for(name) + configurations.fetch name do + raise "Missing configuration for the #{name.inspect} Active Storage service. Configurations available for #{configurations.keys.inspect}" + end + end + + def resolve(class_name) + require "active_storage/service/#{class_name.to_s.underscore}_service" + ActiveStorage::Service.const_get(:"#{class_name.camelize}Service") + rescue LoadError + raise "Missing service adapter for #{class_name.inspect}" + end + end +end diff --git a/activestorage/lib/active_storage/service/disk_service.rb b/activestorage/lib/active_storage/service/disk_service.rb new file mode 100644 index 0000000000..52f3a3df16 --- /dev/null +++ b/activestorage/lib/active_storage/service/disk_service.rb @@ -0,0 +1,162 @@ +# frozen_string_literal: true + +require "fileutils" +require "pathname" +require "digest/md5" +require "active_support/core_ext/numeric/bytes" + +module ActiveStorage + # Wraps a local disk path as an Active Storage service. See ActiveStorage::Service for the generic API + # documentation that applies to all services. + class Service::DiskService < Service + attr_reader :root + + def initialize(root:) + @root = root + end + + def upload(key, io, checksum: nil) + instrument :upload, key: key, checksum: checksum do + IO.copy_stream(io, make_path_for(key)) + ensure_integrity_of(key, checksum) if checksum + end + end + + def download(key, &block) + if block_given? + instrument :streaming_download, key: key do + stream key, &block + end + else + instrument :download, key: key do + begin + File.binread path_for(key) + rescue Errno::ENOENT + raise ActiveStorage::FileNotFoundError + end + end + end + end + + def download_chunk(key, range) + instrument :download_chunk, key: key, range: range do + begin + File.open(path_for(key), "rb") do |file| + file.seek range.begin + file.read range.size + end + rescue Errno::ENOENT + raise ActiveStorage::FileNotFoundError + end + end + end + + def delete(key) + instrument :delete, key: key do + begin + File.delete path_for(key) + rescue Errno::ENOENT + # Ignore files already deleted + end + end + end + + def delete_prefixed(prefix) + instrument :delete_prefixed, prefix: prefix do + Dir.glob(path_for("#{prefix}*")).each do |path| + FileUtils.rm_rf(path) + end + end + end + + def exist?(key) + instrument :exist, key: key do |payload| + answer = File.exist? path_for(key) + payload[:exist] = answer + answer + end + end + + def url(key, expires_in:, filename:, disposition:, content_type:) + instrument :url, key: key do |payload| + verified_key_with_expiration = ActiveStorage.verifier.generate(key, expires_in: expires_in, purpose: :blob_key) + + generated_url = + url_helpers.rails_disk_service_url( + verified_key_with_expiration, + host: current_host, + filename: filename, + disposition: content_disposition_with(type: disposition, filename: filename), + content_type: content_type + ) + + payload[:url] = generated_url + + generated_url + end + end + + def url_for_direct_upload(key, expires_in:, content_type:, content_length:, checksum:) + instrument :url, key: key do |payload| + verified_token_with_expiration = ActiveStorage.verifier.generate( + { + key: key, + content_type: content_type, + content_length: content_length, + checksum: checksum + }, + { expires_in: expires_in, + purpose: :blob_token } + ) + + generated_url = url_helpers.update_rails_disk_service_url(verified_token_with_expiration, host: current_host) + + payload[:url] = generated_url + + generated_url + end + end + + def headers_for_direct_upload(key, content_type:, **) + { "Content-Type" => content_type } + end + + def path_for(key) #:nodoc: + File.join root, folder_for(key), key + end + + private + def stream(key) + File.open(path_for(key), "rb") do |file| + while data = file.read(5.megabytes) + yield data + end + end + rescue Errno::ENOENT + raise ActiveStorage::FileNotFoundError + end + + def folder_for(key) + [ key[0..1], key[2..3] ].join("/") + end + + def make_path_for(key) + path_for(key).tap { |path| FileUtils.mkdir_p File.dirname(path) } + end + + def ensure_integrity_of(key, checksum) + unless Digest::MD5.file(path_for(key)).base64digest == checksum + delete key + raise ActiveStorage::IntegrityError + end + end + + def url_helpers + @url_helpers ||= Rails.application.routes.url_helpers + end + + def current_host + ActiveStorage::Current.host + end + end +end diff --git a/activestorage/lib/active_storage/service/gcs_service.rb b/activestorage/lib/active_storage/service/gcs_service.rb new file mode 100644 index 0000000000..18c0f14cfc --- /dev/null +++ b/activestorage/lib/active_storage/service/gcs_service.rb @@ -0,0 +1,143 @@ +# frozen_string_literal: true + +gem "google-cloud-storage", "~> 1.11" +require "google/cloud/storage" + +module ActiveStorage + # Wraps the Google Cloud Storage as an Active Storage service. See ActiveStorage::Service for the generic API + # documentation that applies to all services. + class Service::GCSService < Service + def initialize(**config) + @config = config + end + + def upload(key, io, checksum: nil) + instrument :upload, key: key, checksum: checksum do + begin + # The official GCS client library doesn't allow us to create a file with no Content-Type metadata. + # We need the file we create to have no Content-Type so we can control it via the response-content-type + # param in signed URLs. Workaround: let the GCS client create the file with an inferred + # Content-Type (usually "application/octet-stream") then clear it. + bucket.create_file(io, key, md5: checksum).update do |file| + file.content_type = nil + end + rescue Google::Cloud::InvalidArgumentError + raise ActiveStorage::IntegrityError + end + end + end + + def download(key, &block) + if block_given? + instrument :streaming_download, key: key do + stream(key, &block) + end + else + instrument :download, key: key do + begin + file_for(key).download.string + rescue Google::Cloud::NotFoundError + raise ActiveStorage::FileNotFoundError + end + end + end + end + + def download_chunk(key, range) + instrument :download_chunk, key: key, range: range do + begin + file_for(key).download(range: range).string + rescue Google::Cloud::NotFoundError + raise ActiveStorage::FileNotFoundError + end + end + end + + def delete(key) + instrument :delete, key: key do + begin + file_for(key).delete + rescue Google::Cloud::NotFoundError + # Ignore files already deleted + end + end + end + + def delete_prefixed(prefix) + instrument :delete_prefixed, prefix: prefix do + bucket.files(prefix: prefix).all do |file| + begin + file.delete + rescue Google::Cloud::NotFoundError + # Ignore concurrently-deleted files + end + end + end + end + + def exist?(key) + instrument :exist, key: key do |payload| + answer = file_for(key).exists? + payload[:exist] = answer + answer + end + end + + def url(key, expires_in:, filename:, content_type:, disposition:) + instrument :url, key: key do |payload| + generated_url = file_for(key).signed_url expires: expires_in, query: { + "response-content-disposition" => content_disposition_with(type: disposition, filename: filename), + "response-content-type" => content_type + } + + payload[:url] = generated_url + + generated_url + end + end + + def url_for_direct_upload(key, expires_in:, checksum:, **) + instrument :url, key: key do |payload| + generated_url = bucket.signed_url key, method: "PUT", expires: expires_in, content_md5: checksum + + payload[:url] = generated_url + + generated_url + end + end + + def headers_for_direct_upload(key, checksum:, **) + { "Content-MD5" => checksum } + end + + private + attr_reader :config + + def file_for(key, skip_lookup: true) + bucket.file(key, skip_lookup: skip_lookup) + end + + # Reads the file for the given key in chunks, yielding each to the block. + def stream(key) + file = file_for(key, skip_lookup: false) + + chunk_size = 5.megabytes + offset = 0 + + raise ActiveStorage::FileNotFoundError unless file.present? + + while offset < file.size + yield file.download(range: offset..(offset + chunk_size - 1)).string + offset += chunk_size + end + end + + def bucket + @bucket ||= client.bucket(config.fetch(:bucket)) + end + + def client + @client ||= Google::Cloud::Storage.new(config.except(:bucket)) + end + end +end diff --git a/activestorage/lib/active_storage/service/mirror_service.rb b/activestorage/lib/active_storage/service/mirror_service.rb new file mode 100644 index 0000000000..6002ef5a00 --- /dev/null +++ b/activestorage/lib/active_storage/service/mirror_service.rb @@ -0,0 +1,55 @@ +# frozen_string_literal: true + +require "active_support/core_ext/module/delegation" + +module ActiveStorage + # Wraps a set of mirror services and provides a single ActiveStorage::Service object that will all + # have the files uploaded to them. A +primary+ service is designated to answer calls to +download+, +exists?+, + # and +url+. + class Service::MirrorService < Service + attr_reader :primary, :mirrors + + delegate :download, :download_chunk, :exist?, :url, to: :primary + + # Stitch together from named services. + def self.build(primary:, mirrors:, configurator:, **options) #:nodoc: + new \ + primary: configurator.build(primary), + mirrors: mirrors.collect { |name| configurator.build name } + end + + def initialize(primary:, mirrors:) + @primary, @mirrors = primary, mirrors + end + + # Upload the +io+ to the +key+ specified to all services. If a +checksum+ is provided, all services will + # ensure a match when the upload has completed or raise an ActiveStorage::IntegrityError. + def upload(key, io, checksum: nil) + each_service.collect do |service| + service.upload key, io.tap(&:rewind), checksum: checksum + end + end + + # Delete the file at the +key+ on all services. + def delete(key) + perform_across_services :delete, key + end + + # Delete files at keys starting with the +prefix+ on all services. + def delete_prefixed(prefix) + perform_across_services :delete_prefixed, prefix + end + + private + def each_service(&block) + [ primary, *mirrors ].each(&block) + end + + def perform_across_services(method, *args) + # FIXME: Convert to be threaded + each_service.collect do |service| + service.public_send method, *args + end + end + end +end diff --git a/activestorage/lib/active_storage/service/s3_service.rb b/activestorage/lib/active_storage/service/s3_service.rb new file mode 100644 index 0000000000..89a9e54158 --- /dev/null +++ b/activestorage/lib/active_storage/service/s3_service.rb @@ -0,0 +1,122 @@ +# frozen_string_literal: true + +require "aws-sdk-s3" +require "active_support/core_ext/numeric/bytes" + +module ActiveStorage + # Wraps the Amazon Simple Storage Service (S3) as an Active Storage service. + # See ActiveStorage::Service for the generic API documentation that applies to all services. + class Service::S3Service < Service + attr_reader :client, :bucket, :upload_options + + def initialize(bucket:, upload: {}, **options) + @client = Aws::S3::Resource.new(**options) + @bucket = @client.bucket(bucket) + + @upload_options = upload + end + + def upload(key, io, checksum: nil) + instrument :upload, key: key, checksum: checksum do + begin + object_for(key).put(upload_options.merge(body: io, content_md5: checksum)) + rescue Aws::S3::Errors::BadDigest + raise ActiveStorage::IntegrityError + end + end + end + + def download(key, &block) + if block_given? + instrument :streaming_download, key: key do + stream(key, &block) + end + else + instrument :download, key: key do + begin + object_for(key).get.body.string.force_encoding(Encoding::BINARY) + rescue Aws::S3::Errors::NoSuchKey + raise ActiveStorage::FileNotFoundError + end + end + end + end + + def download_chunk(key, range) + instrument :download_chunk, key: key, range: range do + begin + object_for(key).get(range: "bytes=#{range.begin}-#{range.exclude_end? ? range.end - 1 : range.end}").body.read.force_encoding(Encoding::BINARY) + rescue Aws::S3::Errors::NoSuchKey + raise ActiveStorage::FileNotFoundError + end + end + end + + def delete(key) + instrument :delete, key: key do + object_for(key).delete + end + end + + def delete_prefixed(prefix) + instrument :delete_prefixed, prefix: prefix do + bucket.objects(prefix: prefix).batch_delete! + end + end + + def exist?(key) + instrument :exist, key: key do |payload| + answer = object_for(key).exists? + payload[:exist] = answer + answer + end + end + + def url(key, expires_in:, filename:, disposition:, content_type:) + instrument :url, key: key do |payload| + generated_url = object_for(key).presigned_url :get, expires_in: expires_in.to_i, + response_content_disposition: content_disposition_with(type: disposition, filename: filename), + response_content_type: content_type + + payload[:url] = generated_url + + generated_url + end + end + + def url_for_direct_upload(key, expires_in:, content_type:, content_length:, checksum:) + instrument :url, key: key do |payload| + generated_url = object_for(key).presigned_url :put, expires_in: expires_in.to_i, + content_type: content_type, content_length: content_length, content_md5: checksum + + payload[:url] = generated_url + + generated_url + end + end + + def headers_for_direct_upload(key, content_type:, checksum:, **) + { "Content-Type" => content_type, "Content-MD5" => checksum } + end + + private + def object_for(key) + bucket.object(key) + end + + # Reads the object for the given key in chunks, yielding each to the block. + def stream(key) + object = object_for(key) + + chunk_size = 5.megabytes + offset = 0 + + raise ActiveStorage::FileNotFoundError unless object.exists? + + while offset < object.content_length + yield object.get(range: "bytes=#{offset}-#{offset + chunk_size - 1}").body.read.force_encoding(Encoding::BINARY) + offset += chunk_size + end + end + end +end |