//! The S3 backend: plain HTTPS with SigV4 signing against any S3 endpoint.
//!
//! Only the two conditional writes matter for correctness — `If-None-Match: *`
//! for objects and saves, `If-Match` for `head` — because they are what stops
//! two machines from silently overwriting each other. Everything else is
//! ordinary GET, DELETE and ListObjectsV2.
module remote
import net.http
import time
pub const s3_service = 's3'
pub const retry_attempts = 3
pub struct S3 {
pub:
endpoint string
bucket string
region string
access_key string
secret_key string
mut:
signer Signer
root string // scheme://host[:port][/base path]
}
pub fn new_s3(endpoint string, bucket string, region string, access_key string, secret_key string) !S3 {
if endpoint == '' || bucket == '' || region == '' {
return error('the S3 remote needs an endpoint, a bucket and a region')
}
if access_key == '' || secret_key == '' {
return error('the S3 remote needs an access key and a secret key')
}
if !endpoint.starts_with('http://') && !endpoint.starts_with('https://') {
return error('the S3 endpoint should start with https:// — got "${endpoint}"')
}
root := endpoint.trim_right('/')
return S3{
endpoint: root
bucket: bucket
region: region
access_key: access_key
secret_key: secret_key
signer: Signer{
access_key: access_key
secret_key: secret_key
region: region
service: s3_service
sign_payload: true
}
root: root
}
}
pub fn (mut s S3) get(key string) !Object {
response := s.send('GET', key, '', map[string]string{}, []u8{})!
match response.status_code {
404 {
return Object{}
}
200 {
return Object{
found: true
body: response.body.bytes()
tag: clean_etag(header_of(response, 'ETag'))
}
}
else {
return error('s3 GET ${key} failed: ${response.status_code} ${short_body(response)}')
}
}
}
pub fn (mut s S3) list(prefix string) ![]string {
mut keys := []string{}
mut token := ''
for {
// The query is passed raw; url_for and the signer encode it the same
// way, which is what keeps the URL and the signature in step.
mut query := 'list-type=2&prefix=${prefix}'
if token != '' {
query += '&continuation-token=${token}'
}
response := s.send('GET', '', query, map[string]string{}, []u8{})!
if response.status_code != 200 {
return error('s3 list failed: ${response.status_code} ${short_body(response)}')
}
keys << elements(response.body, 'Key')
if !response.body.contains('true') {
break
}
token = first_element(response.body, 'NextContinuationToken') or { break }
}
return keys
}
pub fn (mut s S3) put_if_absent(key string, body []u8) !bool {
response := s.send('PUT', key, '', {
'If-None-Match': '*'
}, body)!
if response.status_code == 412 {
return false
}
if !success(response.status_code) {
return error('s3 PUT ${key} failed: ${response.status_code} ${short_body(response)}')
}
return true
}
pub fn (mut s S3) put_if_match(key string, body []u8, tag string) !bool {
mut headers := map[string]string{}
if tag == '' {
// The key must not exist yet.
headers['If-None-Match'] = '*'
} else {
headers['If-Match'] = tag
}
response := s.send('PUT', key, '', headers, body)!
if response.status_code == 412 {
return false
}
if !success(response.status_code) {
return error('s3 PUT ${key} failed: ${response.status_code} ${short_body(response)}')
}
return true
}
pub fn (mut s S3) delete(key string) ! {
response := s.send('DELETE', key, '', map[string]string{}, []u8{})!
if !success(response.status_code) && response.status_code != 404 {
return error('s3 DELETE ${key} failed: ${response.status_code} ${short_body(response)}')
}
}
// send signs one request and retries the failures worth retrying: a network
// error, or a 5xx. A 412 is an answer, not a fault, and is never retried.
fn (mut s S3) send(method string, key string, query string, headers map[string]string, body []u8) !http.Response {
url := s.url_for(key, query)
mut last := ''
for attempt in 1 .. retry_attempts + 1 {
signed := s.signer.sign(method, url, query, headers, body, time.now().local_to_utc())
mut outbound := http.new_header()
for name, value in signed {
outbound.add_custom(header_case(name), value) or {}
}
response := http.fetch(
url: url
method: method_from(method)
header: outbound
data: body.bytestr()
) or {
last = err.msg()
time.sleep(backoff(attempt))
continue
}
if response.status_code >= 500 {
last = '${response.status_code}'
time.sleep(backoff(attempt))
continue
}
return response
}
return error('s3 ${method} ${key} failed after ${retry_attempts} attempts: ${last}')
}
// url_for builds the path-style URL. The path is encoded here exactly as the
// signer encodes it for the canonical request.
fn (s &S3) url_for(key string, query string) string {
mut url := '${s.root}/${encode(s.bucket, false)}'
if key != '' {
url += '/${encode(key, true)}'
}
if query != '' {
url += '?${canonical_query(query)}'
}
return url
}
fn backoff(attempt int) time.Duration {
return time.Duration(attempt) * 200 * time.millisecond
}
fn method_from(method string) http.Method {
return match method {
'GET' { .get }
'PUT' { .put }
'POST' { .post }
'DELETE' { .delete }
'HEAD' { .head }
else { .get }
}
}
fn success(status int) bool {
return status >= 200 && status < 300
}
fn header_of(response http.Response, name string) string {
return response.header.get_custom(name, http.HeaderQueryConfig{}) or { '' }
}
// clean_etag strips the quotes S3 puts around an ETag: the value is sent back
// verbatim in `If-Match`, and a quoted form would not match.
pub fn clean_etag(value string) string {
return value.trim_space().trim('"')
}
fn header_case(name string) string {
return match name {
'authorization' { 'Authorization' }
'host' { 'Host' }
'x-amz-date' { 'X-Amz-Date' }
'x-amz-content-sha256' { 'X-Amz-Content-Sha256' }
'if-none-match' { 'If-None-Match' }
'if-match' { 'If-Match' }
else { name }
}
}
fn short_body(response http.Response) string {
body := response.body
if body.len <= 200 {
return body
}
return body[..200]
}
// elements pulls every text between and out of a response. The
// only response Koh reads is S3's ListObjectsV2, which is flat and small, so a
// scanner is honest here where a full XML parser would be a dependency.
pub fn elements(xml string, name string) []string {
open := '<${name}>'
close := '${name}>'
mut out := []string{}
mut rest := xml
for {
start := rest.index(open) or { break }
after := rest[start + open.len..]
end := after.index(close) or { break }
out << unescape(after[..end])
rest = after[end + close.len..]
}
return out
}
fn first_element(xml string, name string) ?string {
values := elements(xml, name)
if values.len == 0 {
return none
}
return values[0]
}
fn unescape(text string) string {
return text.replace('&', '&').replace('<', '<').replace('>', '>').replace('"',
'"').replace(''', "'")
}