aboutsummaryrefslogtreecommitdiffstats
path: root/net
diff options
context:
space:
mode:
Diffstat (limited to 'net')
-rw-r--r--net/connect.lua26
-rw-r--r--net/resolvers/basic.lua160
-rw-r--r--net/resolvers/service.lua81
3 files changed, 168 insertions, 99 deletions
diff --git a/net/connect.lua b/net/connect.lua
index 4b602be4..241cc65b 100644
--- a/net/connect.lua
+++ b/net/connect.lua
@@ -1,6 +1,7 @@
local server = require "net.server";
local log = require "util.logger".init("net.connect");
local new_id = require "util.id".short;
+local timer = require "util.timer";
-- TODO #1246 Happy Eyeballs
-- FIXME RFC 6724
@@ -28,11 +29,7 @@ local pending_connection_listeners = {};
local function attempt_connection(p)
p:log("debug", "Checking for targets...");
- if p.conn then
- pending_connections_map[p.conn] = nil;
- p.conn = nil;
- end
- p.target_resolver:next(function (conn_type, ip, port, extra)
+ p.target_resolver:next(function (conn_type, ip, port, extra, more_targets_available)
if not conn_type then
-- No more targets to try
p:log("debug", "No more connection targets to try", p.target_resolver.last_error);
@@ -49,8 +46,16 @@ local function attempt_connection(p)
p.last_error = err or "unknown reason";
return attempt_connection(p);
end
- p.conn = conn;
+ p.conns[conn] = true;
pending_connections_map[conn] = p;
+ if more_targets_available then
+ timer.add_task(0.250, function ()
+ if not p.connected then
+ p:log("debug", "Still not connected, making parallel connection attempt...");
+ attempt_connection(p);
+ end
+ end);
+ end
end);
end
@@ -62,6 +67,13 @@ function pending_connection_listeners.onconnect(conn)
return;
end
pending_connections_map[conn] = nil;
+ if p.connected then
+ -- We already succeeded in connecting
+ p.conns[conn] = nil;
+ conn:close();
+ return;
+ end
+ p.connected = true;
p:log("debug", "Successfully connected");
conn:setlistener(p.listeners, p.data);
return p.listeners.onconnect(conn);
@@ -73,6 +85,7 @@ function pending_connection_listeners.ondisconnect(conn, reason)
log("warn", "Failed connection, but unexpected!");
return;
end
+ p.conns[conn] = nil;
p.last_error = reason or "unknown reason";
p:log("debug", "Connection attempt failed: %s", p.last_error);
attempt_connection(p);
@@ -85,6 +98,7 @@ local function connect(target_resolver, listeners, options, data)
listeners = assert(listeners);
options = options or {};
data = data;
+ conns = {};
}, pending_connection_mt);
p:log("debug", "Starting connection process");
diff --git a/net/resolvers/basic.lua b/net/resolvers/basic.lua
index 305bce76..15338ff4 100644
--- a/net/resolvers/basic.lua
+++ b/net/resolvers/basic.lua
@@ -2,13 +2,59 @@ local adns = require "net.adns";
local inet_pton = require "util.net".pton;
local inet_ntop = require "util.net".ntop;
local idna_to_ascii = require "util.encodings".idna.to_ascii;
-local unpack = table.unpack or unpack; -- luacheck: ignore 113
+local promise = require "util.promise";
+local t_move = require "util.table".move;
local methods = {};
local resolver_mt = { __index = methods };
-- FIXME RFC 6724
+local function do_dns_lookup(self, dns_resolver, record_type, name)
+ return promise.new(function (resolve, reject)
+ local ipv = (record_type == "A" and "4") or (record_type == "AAAA" and "6") or nil;
+ if ipv and self.extra["use_ipv"..ipv] == false then
+ return reject(("IPv%s disabled - %s lookup skipped"):format(ipv, record_type));
+ elseif record_type == "TLSA" and self.extra.use_dane ~= true then
+ return reject("DANE disabled - TLSA lookup skipped");
+ end
+ dns_resolver:lookup(function (answer, err)
+ if not answer then
+ return reject(err);
+ elseif answer.bogus then
+ return reject(("Validation error in %s lookup"):format(record_type));
+ elseif answer.status and #answer == 0 then
+ return reject(("%s in %s lookup"):format(answer.status, record_type));
+ end
+
+ local targets = { secure = answer.secure };
+ for _, record in ipairs(answer) do
+ if ipv then
+ table.insert(targets, { self.conn_type..ipv, record[record_type:lower()], self.port, self.extra });
+ else
+ table.insert(targets, record[record_type:lower()]);
+ end
+ end
+ return resolve(targets);
+ end, name, record_type, "IN");
+ end);
+end
+
+local function merge_targets(ipv4_targets, ipv6_targets)
+ local result = { secure = ipv4_targets.secure and ipv6_targets.secure };
+ local common_length = math.min(#ipv4_targets, #ipv6_targets);
+ for i = 1, common_length do
+ table.insert(result, ipv6_targets[i]);
+ table.insert(result, ipv4_targets[i]);
+ end
+ if common_length < #ipv4_targets then
+ t_move(ipv4_targets, common_length+1, #ipv4_targets, common_length+1, result);
+ elseif common_length < #ipv6_targets then
+ t_move(ipv6_targets, common_length+1, #ipv6_targets, common_length+1, result);
+ end
+ return result;
+end
+
-- Find the next target to connect to, and
-- pass it to cb()
function methods:next(cb)
@@ -18,7 +64,7 @@ function methods:next(cb)
return;
end
local next_target = table.remove(self.targets, 1);
- cb(unpack(next_target, 1, 4));
+ cb(next_target[1], next_target[2], next_target[3], next_target[4], not not self.targets[1]);
return;
end
@@ -28,91 +74,45 @@ function methods:next(cb)
return;
end
- local secure = true;
- local tlsa = {};
- local targets = {};
- local n = 3;
- local function ready()
- n = n - 1;
- if n > 0 then return; end
- self.targets = targets;
+ -- Resolve DNS to target list
+ local dns_resolver = adns.resolver();
+
+ local dns_lookups = {
+ ipv4 = do_dns_lookup(self, dns_resolver, "A", self.hostname);
+ ipv6 = do_dns_lookup(self, dns_resolver, "AAAA", self.hostname);
+ tlsa = do_dns_lookup(self, dns_resolver, "TLSA", ("_%d._%s.%s"):format(self.port, self.conn_type, self.hostname));
+ };
+
+ promise.all_settled(dns_lookups):next(function (dns_results)
+ -- Combine targets, assign to self.targets, self:next(cb)
+ local have_ipv4 = dns_results.ipv4.status == "fulfilled";
+ local have_ipv6 = dns_results.ipv6.status == "fulfilled";
+
+ if have_ipv4 and have_ipv6 then
+ self.targets = merge_targets(dns_results.ipv4.value, dns_results.ipv6.value);
+ elseif have_ipv4 then
+ self.targets = dns_results.ipv4.value;
+ elseif have_ipv6 then
+ self.targets = dns_results.ipv6.value;
+ else
+ self.targets = {};
+ end
+
if self.extra and self.extra.use_dane then
- if secure and tlsa[1] then
- self.extra.tlsa = tlsa;
+ if self.targets.secure and dns_results.tlsa.status == "fulfilled" then
+ self.extra.tlsa = dns_results.tlsa.value;
self.extra.dane_hostname = self.hostname;
else
self.extra.tlsa = nil;
self.extra.dane_hostname = nil;
end
end
- self:next(cb);
- end
- -- Resolve DNS to target list
- local dns_resolver = adns.resolver();
-
- if not self.extra or self.extra.use_ipv4 ~= false then
- dns_resolver:lookup(function (answer, err)
- if answer then
- secure = secure and answer.secure;
- for _, record in ipairs(answer) do
- table.insert(targets, { self.conn_type.."4", record.a, self.port, self.extra });
- end
- if answer.bogus then
- self.last_error = "Validation error in A lookup";
- elseif answer.status then
- self.last_error = answer.status .. " in A lookup";
- end
- else
- self.last_error = err;
- end
- ready();
- end, self.hostname, "A", "IN");
- else
- ready();
- end
-
- if not self.extra or self.extra.use_ipv6 ~= false then
- dns_resolver:lookup(function (answer, err)
- if answer then
- secure = secure and answer.secure;
- for _, record in ipairs(answer) do
- table.insert(targets, { self.conn_type.."6", record.aaaa, self.port, self.extra });
- end
- if answer.bogus then
- self.last_error = "Validation error in AAAA lookup";
- elseif answer.status then
- self.last_error = answer.status .. " in AAAA lookup";
- end
- else
- self.last_error = err;
- end
- ready();
- end, self.hostname, "AAAA", "IN");
- else
- ready();
- end
-
- if self.extra and self.extra.use_dane == true then
- dns_resolver:lookup(function (answer, err)
- if answer then
- secure = secure and answer.secure;
- for _, record in ipairs(answer) do
- table.insert(tlsa, record.tlsa);
- end
- if answer.bogus then
- self.last_error = "Validation error in TLSA lookup";
- elseif answer.status then
- self.last_error = answer.status .. " in TLSA lookup";
- end
- else
- self.last_error = err;
- end
- ready();
- end, ("_%d._tcp.%s"):format(self.port, self.hostname), "TLSA", "IN");
- else
- ready();
- end
+ self:next(cb);
+ end):catch(function (err)
+ self.last_error = err;
+ self.targets = {};
+ end);
end
local function new(hostname, port, conn_type, extra)
@@ -137,7 +137,7 @@ local function new(hostname, port, conn_type, extra)
hostname = ascii_host;
port = port;
conn_type = conn_type;
- extra = extra;
+ extra = extra or {};
targets = targets;
}, resolver_mt);
end
diff --git a/net/resolvers/service.lua b/net/resolvers/service.lua
index 3810cac8..a7ce76a3 100644
--- a/net/resolvers/service.lua
+++ b/net/resolvers/service.lua
@@ -2,23 +2,78 @@ local adns = require "net.adns";
local basic = require "net.resolvers.basic";
local inet_pton = require "util.net".pton;
local idna_to_ascii = require "util.encodings".idna.to_ascii;
-local unpack = table.unpack or unpack; -- luacheck: ignore 113
local methods = {};
local resolver_mt = { __index = methods };
+local function new_target_selector(rrset)
+ local rr_count = rrset and #rrset;
+ if not rr_count or rr_count == 0 then
+ rrset = nil;
+ else
+ table.sort(rrset, function (a, b) return a.srv.priority < b.srv.priority end);
+ end
+ local rrset_pos = 1;
+ local priority_bucket, bucket_total_weight, bucket_len, bucket_used;
+ return function ()
+ if not rrset then return; end
+
+ if not priority_bucket or bucket_used >= bucket_len then
+ if rrset_pos > rr_count then return; end -- Used up all records
+
+ -- Going to start on a new priority now. Gather up all the next
+ -- records with the same priority and add them to priority_bucket
+ priority_bucket, bucket_total_weight, bucket_len, bucket_used = {}, 0, 0, 0;
+ local current_priority;
+ repeat
+ local curr_record = rrset[rrset_pos].srv;
+ if not current_priority then
+ current_priority = curr_record.priority;
+ elseif current_priority ~= curr_record.priority then
+ break;
+ end
+ table.insert(priority_bucket, curr_record);
+ bucket_total_weight = bucket_total_weight + curr_record.weight;
+ bucket_len = bucket_len + 1;
+ rrset_pos = rrset_pos + 1;
+ until rrset_pos > rr_count;
+ end
+
+ bucket_used = bucket_used + 1;
+ local n, running_total = math.random(0, bucket_total_weight), 0;
+ local target_record;
+ for i = 1, bucket_len do
+ local candidate = priority_bucket[i];
+ if candidate then
+ running_total = running_total + candidate.weight;
+ if running_total >= n then
+ target_record = candidate;
+ bucket_total_weight = bucket_total_weight - candidate.weight;
+ priority_bucket[i] = nil;
+ break;
+ end
+ end
+ end
+ return target_record;
+ end;
+end
+
-- Find the next target to connect to, and
-- pass it to cb()
function methods:next(cb)
- if self.targets then
- if not self.resolver then
- if #self.targets == 0 then
+ if self.resolver or self._get_next_target then
+ if not self.resolver then -- Do we have a basic resolver currently?
+ -- We don't, so fetch a new SRV target, create a new basic resolver for it
+ local next_srv_target = self._get_next_target and self._get_next_target();
+ if not next_srv_target then
+ -- No more SRV targets left
cb(nil);
return;
end
- local next_target = table.remove(self.targets, 1);
- self.resolver = basic.new(unpack(next_target, 1, 4));
+ -- Create a new basic resolver for this SRV target
+ self.resolver = basic.new(next_srv_target.target, next_srv_target.port, self.conn_type, self.extra);
end
+ -- Look up the next (basic) target from the current target's resolver
self.resolver:next(function (...)
if self.resolver then
self.last_error = self.resolver.last_error;
@@ -31,6 +86,9 @@ function methods:next(cb)
end
end);
return;
+ elseif self.in_progress then
+ cb(nil);
+ return;
end
if not self.hostname then
@@ -39,9 +97,9 @@ function methods:next(cb)
return;
end
- local targets = {};
+ self.in_progress = true;
+
local function ready()
- self.targets = targets;
self:next(cb);
end
@@ -63,7 +121,7 @@ function methods:next(cb)
if #answer == 0 then
if self.extra and self.extra.default_port then
- table.insert(targets, { self.hostname, self.extra.default_port, self.conn_type, self.extra });
+ self.resolver = basic.new(self.hostname, self.extra.default_port, self.conn_type, self.extra);
else
self.last_error = "zero SRV records found";
end
@@ -77,10 +135,7 @@ function methods:next(cb)
return;
end
- table.sort(answer, function (a, b) return a.srv.priority < b.srv.priority end);
- for _, record in ipairs(answer) do
- table.insert(targets, { record.srv.target, record.srv.port, self.conn_type, self.extra });
- end
+ self._get_next_target = new_target_selector(answer);
else
self.last_error = err;
end
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-22 00:21:57 +0000