Part one of internal component support
[prosody.git] / core / s2smanager.lua
1
2 local hosts = hosts;
3 local sessions = sessions;
4 local socket = require "socket";
5 local format = string.format;
6 local t_insert = table.insert;
7 local tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber
8     = tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber;
9
10 local connlisteners_get = require "net.connlisteners".get;
11 local wraptlsclient = require "net.server".wraptlsclient;
12 local modulemanager = require "core.modulemanager";
13
14 local uuid_gen = require "util.uuid".generate;
15
16 local logger_init = require "util.logger".init;
17
18 local log = logger_init("s2smanager");
19
20 local md5_hash = require "util.hashes".md5;
21
22 local dialback_secret = "This is very secret!!! Ha!";
23
24 local srvmap = { ["gmail.com"] = "talk.google.com", ["identi.ca"] = "longlance.controlezvous.ca", ["cdr.se"] = "jabber.cdr.se" };
25
26 module "s2smanager"
27
28 function connect_host(from_host, to_host)
29 end
30
31 function send_to_host(from_host, to_host, data)
32         local host = hosts[to_host];
33         if host then
34                 -- Write to connection
35                 if host.type == "s2sout_unauthed" and not host.notopen and not host.dialback_key then
36                         log("debug", "trying to send over unauthed s2sout to "..to_host..", authing it now...");
37                         initiate_dialback(host);
38                         if not host.sendq then host.sendq = { data };
39                         else t_insert(host.sendq, data); end
40                 else
41                         log("debug", "going to send stanza to "..to_host.." from "..from_host);
42                         hosts[to_host].sends2s(data);
43                         log("debug", "stanza sent over "..hosts[to_host].type);
44                 end
45         else
46                 log("debug", "opening a new outgoing connection for this stanza");
47                 local host_session = new_outgoing(from_host, to_host);
48                 -- Store in buffer
49                 host_session.sendq = { data };
50         end
51 end
52
53 function disconnect_host(host)
54         
55 end
56
57 local open_sessions = 0;
58
59 function new_incoming(conn)
60         local session = { conn = conn, type = "s2sin_unauthed", direction = "incoming" };
61         if true then
62                 session.trace = newproxy(true);
63                 getmetatable(session.trace).__gc = function () open_sessions = open_sessions - 1; print("s2s session got collected, now "..open_sessions.." s2s sessions are allocated") end;
64         end
65         open_sessions = open_sessions + 1;
66         local w = conn.write;
67         session.sends2s = function (t) w(tostring(t)); end
68         return session;
69 end
70
71 function new_outgoing(from_host, to_host)
72                 local host_session = { to_host = to_host, from_host = from_host, notopen = true, type = "s2sout_unauthed", direction = "outgoing" };
73                 hosts[to_host] = host_session;
74                 local cl = connlisteners_get("xmppserver");
75                 
76                 local conn, handler = socket.tcp()
77                 --FIXME: Below parameters (ports/ip) are incorrect (use SRV)
78                 to_host = srvmap[to_host] or to_host;
79                 conn:connect(to_host, 5269);
80                 conn = wraptlsclient(cl, conn, to_host, 5269, 0, 1, hosts[from_host].ssl_ctx );
81                 host_session.conn = conn;
82                 
83                 -- Register this outgoing connection so that xmppserver_listener knows about it
84                 -- otherwise it will assume it is a new incoming connection
85                 cl.register_outgoing(conn, host_session);
86                 
87                 do
88                         local conn_name = "s2sout"..tostring(conn):match("[a-f0-9]*$");
89                         host_session.log = logger_init(conn_name);
90                 end
91                 
92                 local w = conn.write;
93                 host_session.sends2s = function (t) w(tostring(t)); end
94                 
95                 conn.write(format([[<stream:stream xmlns='jabber:server' xmlns:db='jabber:server:dialback' xmlns:stream='http://etherx.jabber.org/streams' from='%s' to='%s' version='1.0'>]], from_host, to_host));
96                  
97                 return host_session;
98 end
99
100 function streamopened(session, attr)
101         session.log("debug", "s2s stream opened");
102         local send = session.sends2s;
103         
104         session.version = tonumber(attr.version) or 0;
105         if session.version >= 1.0 and not (attr.to and attr.from) then
106                 print("to: "..tostring(attr.to).." from: "..tostring(attr.from));
107                 --error(session.to_host.." failed to specify 'to' or 'from' hostname as per RFC");
108                 log("warn", (session.to_host or "(unknown)").." failed to specify 'to' or 'from' hostname as per RFC");
109         end
110         
111         if session.direction == "incoming" then
112                 -- Send a reply stream header
113                 
114                 for k,v in pairs(attr) do print("", tostring(k), ":::", tostring(v)); end
115                 
116                 session.to_host = attr.to;
117                 session.from_host = attr.from;
118         
119                 session.streamid = uuid_gen();
120                 print(session, session.from_host, "incoming s2s stream opened");
121                 send("<?xml version='1.0'?>");
122                 send(format("<stream:stream xmlns='jabber:server' xmlns:db='jabber:server:dialback' xmlns:stream='http://etherx.jabber.org/streams' id='%s' from='%s'>", session.streamid, session.to_host));
123         elseif session.direction == "outgoing" then
124                 -- If we are just using the connection for verifying dialback keys, we won't try and auth it
125                 if not attr.id then error("stream response did not give us a streamid!!!"); end
126                 session.streamid = attr.id;
127         
128                 if not session.dialback_verifying then
129                         initiate_dialback(session);
130                 else
131                         mark_connected(session);
132                 end
133         end
134         --[[
135         local features = {};
136         modulemanager.fire_event("stream-features-s2s", session, features);
137         
138         send("<stream:features>");
139         
140         for _, feature in ipairs(features) do
141                 send(tostring(feature));
142         end
143
144         send("</stream:features>");]]
145         log("info", "s2s stream opened successfully");
146         session.notopen = nil;
147 end
148
149 function initiate_dialback(session)
150         -- generate dialback key
151         session.dialback_key = generate_dialback(session.streamid, session.to_host, session.from_host);
152         session.sends2s(format("<db:result from='%s' to='%s'>%s</db:result>", session.from_host, session.to_host, session.dialback_key));
153         session.log("info", "sent dialback key on outgoing s2s stream");
154 end
155
156 function generate_dialback(id, to, from)
157         return md5_hash(id..to..from..dialback_secret); -- FIXME: See XEP-185 and XEP-220
158 end
159
160 function verify_dialback(id, to, from, key)
161         return key == generate_dialback(id, to, from);
162 end
163
164 function make_authenticated(session)
165         if session.type == "s2sout_unauthed" then
166                 session.type = "s2sout";
167         elseif session.type == "s2sin_unauthed" then
168                 session.type = "s2sin";
169         else
170                 return false;
171         end
172         session.log("info", "connection is now authenticated");
173         
174         mark_connected(session);
175         
176         return true;
177 end
178
179 function mark_connected(session)
180         local sendq, send = session.sendq, session.sends2s;
181         
182         local from, to = session.from_host, session.to_host;
183         
184         session.log("debug", session.direction.." s2s connection "..from.."->"..to.." is now complete");
185         
186         local send_to_host = send_to_host;
187         function session.send(data) send_to_host(to, from, data); end
188         
189         
190         if session.direction == "outgoing" then
191                 if sendq then
192                         session.log("debug", "sending queued stanzas across new outgoing connection to "..session.to_host);
193                         for i, data in ipairs(sendq) do
194                                 send(data);
195                                 sendq[i] = nil;
196                         end
197                         session.sendq = nil;
198                 end
199         end
200 end
201
202 function destroy_session(session)
203         (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host));
204         if session.direction == "outgoing" then
205                 hosts[session.to_host] = nil;
206         end
207         session.conn = nil;
208         session.disconnect = nil;
209         for k in pairs(session) do
210                 if k ~= "trace" then
211                         session[k] = nil;
212                 end
213         end
214 end
215
216 return _M;