local http = require "prosody.net.http";
local array = require "prosody.util.array";
local async = require "prosody.util.async";
local dt = require "prosody.util.datetime";
local hashes = require "prosody.util.hashes";
local httputil = require "prosody.util.http";
local it = require "prosody.util.iterators";
local jid = require "prosody.util.jid";
local json = require "prosody.util.json";
local promise = require "prosody.util.promise";
local set = require "prosody.util.set";
local st = require "prosody.util.stanza";
local uuid = require "prosody.util.uuid";
local xml = require "prosody.util.xml";
local url = require "socket.url";
local new_uuid = uuid.v7 or uuid.generate;
local hmac_sha256 = hashes.hmac_sha256;
local sha256 = hashes.sha256;
local driver = {};
local bucket = module:get_option_string("s3_bucket", "prosody");
local base_uri = module:get_option_string("s3_base_uri", "http://localhost:9000");
local region = module:get_option_string("s3_region", "us-east-1");
local access_key = module:get_option_string("s3_access_key");
local secret_key = module:get_option_string("s3_secret_key");
local aws4_format = "AWS4-HMAC-SHA256 Credential=%s/%s, SignedHeaders=%s, Signature=%s";
local function aws_auth(event)
local request, options = event.request, event.options;
local method = options.method or "GET";
local query = options.query;
local payload = options.body;
local payload_type = nil;
if st.is_stanza(payload) then
payload_type = "application/xml";
payload = tostring(payload);
elseif payload ~= nil then
payload_type = "application/json";
payload = json.encode(payload);
end
options.body = payload;
local payload_hash = sha256(payload or "", true);
local now = os.time();
local aws_datetime = os.date("!%Y%m%dT%H%M%SZ", now);
local aws_date = os.date("!%Y%m%d", now);
local headers = {
["Accept"] = "*/*";
["Authorization"] = nil;
["Content-Type"] = payload_type;
["Host"] = request.authority;
["User-Agent"] = "Prosody XMPP Server";
["X-Amz-Content-Sha256"] = payload_hash;
["X-Amz-Date"] = aws_datetime;
};
local canonical_uri = url.build({ path = request.path });
local canonical_query = "";
local canonical_headers = array();
local signed_headers = array()
if query then
local sorted_query = array();
for name, value in it.sorted_pairs(query) do
sorted_query:push({ name = name; value = value });
end
sorted_query:sort(function (a,b) return a.name < b.name end)
canonical_query = httputil.formencode(sorted_query):gsub("%%%x%x", string.upper);
request.query = canonical_query;
end
for header_name, header_value in it.sorted_pairs(headers) do
header_name = header_name:lower();
canonical_headers:push(header_name .. ":" .. header_value .. "\n");
signed_headers:push(header_name);
end
canonical_headers = canonical_headers:concat();
signed_headers = signed_headers:concat(";");
local scope = aws_date .. "/" .. region .. "/s3/aws4_request";
local canonical_request = method .. "\n"
.. canonical_uri .. "\n"
.. canonical_query .. "\n"
.. canonical_headers .. "\n"
.. signed_headers .. "\n"
.. payload_hash;
local signature_payload = "AWS4-HMAC-SHA256" .. "\n" .. aws_datetime .. "\n" .. scope .. "\n" .. sha256(canonical_request, true);
-- This can be cached?
local date_key = hmac_sha256("AWS4" .. secret_key, aws_date);
local date_region_key = hmac_sha256(date_key, region);
local date_region_service_key = hmac_sha256(date_region_key, "s3");
local signing_key = hmac_sha256(date_region_service_key, "aws4_request");
local signature = hmac_sha256(signing_key, signature_payload, true);
headers["Authorization"] = string.format(aws4_format, access_key, scope, signed_headers, signature);
options.headers = headers;
end
function driver:open(store, typ)
local mt = self[typ or "keyval"]
if not mt then
return nil, "unsupported-store";
end
local httpclient = http.new({ connection_pooling = true });
httpclient.events.add_handler("pre-request", aws_auth);
return setmetatable({ store = store; bucket = bucket; type = typ; http = httpclient }, mt);
end
local keyval = { };
driver.keyval = { __index = keyval; __name = module.name .. " keyval store" };
local function new_request(self, method, path, query, payload)
local request = url.parse(base_uri);
request.path = path;
return self.http:request(url.build(request), { method = method; body = payload; query = query });
end
-- coerce result back into Prosody data type
local function on_result(response)
if response.code == 404 and response.request.method == "GET" then
return nil;
end
if response.code >= 400 then
error(response.body);
end
local content_type = response.headers["content-type"];
if content_type == "application/json" then
return json.decode(response.body);
elseif content_type == "application/xml" then
return xml.parse(response.body);
elseif content_type == "application/x-www-form-urlencoded" then
return httputil.formdecode(response.body);
else
module:log("warn", "Unknown response data type %s", content_type);
return response.body;
end
end
function keyval:_path(key)
return url.build_path({
is_absolute = true;
bucket;
jid.escape(module.host);
jid.escape(key or "@");
jid.escape(self.store);
})
end
function keyval:get(user)
return async.wait_for(new_request(self, "GET", self:_path(user)):next(on_result));
end
function keyval:set(user, data)
if data == nil or (type(data) == "table" and next(data) == nil) then
return async.wait_for(new_request(self, "DELETE", self:_path(user)));
end
return async.wait_for(new_request(self, "PUT", self:_path(user), nil, data));
end
function keyval:users()
local bucket_path = url.build_path({ is_absolute = true; bucket; is_directory = true });
local prefix = jid.escape(module.host) .. "/";
local list_result, err = async.wait_for(new_request(self, "GET", bucket_path, { prefix = prefix }))
if err or list_result.code ~= 200 then
return nil, err;
end
local list_bucket_result = xml.parse(list_result.body);
if list_bucket_result:get_child_text("IsTruncated") == "true" then
local max_keys = list_bucket_result:get_child_text("MaxKeys");
module:log("warn", "Paging truncated results not implemented, max %s %s returned", max_keys, self.store);
end
local keys = array();
local store_part = jid.escape(self.store);
for content in list_bucket_result:childtags("Contents") do
local key = url.parse_path(content:get_child_text("Key"));
if key[3] == store_part then
keys:push(jid.unescape(key[2]));
end
end
return function()
return keys:pop();
end
end
local archive = {};
driver.archive = { __index = archive };
archive.caps = {
full_id_range = true; -- both before and after used
ids = true;
};
function archive:_path(username, date, when, with, key)
return url.build_path({
is_absolute = true;
bucket;
jid.escape(module.host);
jid.escape(username or "@");
jid.escape(self.store);
date or dt.date(when);
jid.escape(with and jid.prep(with) or "@");
key;
})
end
-- PUT .../with/when/id
function archive:append(username, key, value, when, with)
key = key or new_uuid();
return async.wait_for(new_request(self, "PUT", self:_path(username, nil, when, with, key), nil, value):next(function(r)
if r.code == 200 then
return key;
else
error(r.body);
end
end));
end
function archive:find(username, query)
local bucket_path = url.build_path({ is_absolute = true; bucket; is_directory = true });
local prefix = { jid.escape(module.host); jid.escape(username or "@"); jid.escape(self.store) };
if not query then
query = {};
end
if query["start"] and query["end"] and dt.date(query["start"]) == dt.date(query["end"]) then
table.insert(prefix, dt.date(query["start"]));
if query["with"] then
table.insert(prefix, jid.escape(query["with"]));
end
end
prefix = table.concat(prefix, "/").."/";
local list_result, err = async.wait_for(new_request(self, "GET", bucket_path, {
prefix = prefix;
["max-keys"] = query["limit"] and tostring(query["limit"]);
}));
if err or list_result.code ~= 200 then
return nil, err;
end
local list_bucket_result = xml.parse(list_result.body);
if list_bucket_result:get_child_text("IsTruncated") == "true" then
local max_keys = list_bucket_result:get_child_text("MaxKeys");
module:log("warn", "Paging truncated results not implemented, max %s %s returned", max_keys, self.store);
end
local keys = array();
local iterwrap = function(...)
return ...;
end
if query["reverse"] then
query["before"], query["after"] = query["after"], query["before"];
iterwrap = it.reverse;
end
local ids = query["ids"] and set.new(query["ids"]);
local found = not query["after"];
for content in iterwrap(list_bucket_result:childtags("Contents")) do
local date, with, id = table.unpack(url.parse_path(content:get_child_text("Key")), 4);
local when = dt.parse(content:get_child_text("LastModified"));
with = jid.unescape(with);
if found and query["before"] == id then
break
end
if (not query["with"] or query["with"] == with)
and (not query["start"] or query["start"] <= when)
and (not query["end"] or query["end"] >= when)
and (not ids or ids:contains(id))
and found then
keys:push({ key = id; date = date; when = when; with = with });
end
if not found and id == query["after"] then
found = not found
end
end
keys:sort(function(a, b)
if a.date ~= b.date then
return a.date < b.date
end
if a.when ~= b.when then
return a.when < b.when;
end
return a.key < b.key;
end);
if query["reverse"] then
keys:reverse();
end
local i = 0;
local function get_next()
i = i + 1;
local item = keys[i];
if item == nil then
return nil;
end
-- luacheck: ignore 431/err
local value, err = async.wait_for(new_request(self, "GET", self:_path(username or "@", item.date, nil, item.with, item.key)):next(on_result));
if not value then
module:log("error", "%s", err);
return nil;
end
return item.key, value, item.when, item.with;
end
return get_next;
end
function archive:users()
return it.unique(keyval.users(self));
end
local function count(t) local n = 0; for _ in pairs(t) do n = n + 1; end return n; end
function archive:delete(username, query)
local deletions = {};
for key, _, when, with in self:find(username, query) do
deletions[key] = new_request(self, "DELETE", self:_path(username or "@", dt.date(when), nil, with, key));
end
return async.wait_for(promise.all(deletions):next(count));
end
module:provides("storage", driver);