anotherim/anotherim-desktop
2
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Merge pull request #592 from hrxi/gsoc_1

Browse source 
Add SOCKS5 bytestreams and a couple of other fixes
This commit is contained in:
Marvin W 2019-08-29 00:44:59 +02:00 committed by GitHub
parent f0c2ce0047 6028fd15a8
commit 9f613d5812
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 1369 additions and 177 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
libdino/src/service/file_manager.vala
View file

@ -269,8 +269,7 @@ public class FileManager : StreamInteractionModule, Object {
}
OutputStream os = file.create(FileCreateFlags.REPLACE_DESTINATION);
yield os.splice_async(input_stream, 0);
os.close();
yield os.splice_async(input_stream, OutputStreamSpliceFlags.CLOSE_SOURCE|OutputStreamSpliceFlags.CLOSE_TARGET);
file_transfer.size = (int)file_meta.size;
file_transfer.file_name = file_meta.file_name;
file_transfer.path = file.get_basename();
@ -327,8 +326,7 @@ public class FileManager : StreamInteractionModule, Object {
string filename = Random.next_int().to_string("%x") + "_" + file_transfer.file_name;
File file = File.new_for_path(Path.build_filename(get_storage_dir(), filename));
OutputStream os = file.create(FileCreateFlags.REPLACE_DESTINATION);
yield os.splice_async(file_transfer.input_stream, 0);
os.close();
yield os.splice_async(file_transfer.input_stream, OutputStreamSpliceFlags.CLOSE_SOURCE|OutputStreamSpliceFlags.CLOSE_TARGET);
file_transfer.state = FileTransfer.State.COMPLETE;
file_transfer.path = filename;
file_transfer.input_stream = yield file.read_async();

4
libdino/src/service/jingle_file_transfers.vala
View file

@ -39,7 +39,11 @@ public class JingleFileProvider : FileProvider, Object {
if (jingle_file_transfer == null) {
throw new FileReceiveError.DOWNLOAD_FAILED("Transfer data not available anymore");
}
try {
jingle_file_transfer.accept(stream);
} catch (IOError e) {
throw new FileReceiveError.DOWNLOAD_FAILED("Establishing connection did not work");
}
return jingle_file_transfer.stream;
}

2
libdino/src/service/module_manager.vala
View file

@ -78,8 +78,10 @@ public class ModuleManager {
module_map[account].add(new StreamError.Module());
module_map[account].add(new Xep.InBandRegistration.Module());
module_map[account].add(new Xep.HttpFileUpload.Module());
module_map[account].add(new Xep.Socks5Bytestreams.Module());
module_map[account].add(new Xep.InBandBytestreams.Module());
module_map[account].add(new Xep.Jingle.Module());
module_map[account].add(new Xep.JingleSocks5Bytestreams.Module());
module_map[account].add(new Xep.JingleInBandBytestreams.Module());
module_map[account].add(new Xep.JingleFileTransfer.Module());
initialize_account_modules(account, module_map[account]);

2
xmpp-vala/CMakeLists.txt
View file

@ -55,6 +55,7 @@ SOURCES
"src/module/xep/0049_private_xml_storage.vala"
"src/module/xep/0054_vcard/module.vala"
"src/module/xep/0060_pubsub.vala"
"src/module/xep/0065_socks5_bytestreams.vala"
"src/module/xep/0066_out_of_band_data.vala"
"src/module/xep/0077_in_band_registration.vala"
"src/module/xep/0082_date_time_profiles.vala"
@ -68,6 +69,7 @@ SOURCES
"src/module/xep/0199_ping.vala"
"src/module/xep/0203_delayed_delivery.vala"
"src/module/xep/0234_jingle_file_transfer.vala"
"src/module/xep/0260_jingle_socks5_bytestreams.vala"
"src/module/xep/0261_jingle_in_band_bytestreams.vala"
"src/module/xep/0280_message_carbons.vala"
"src/module/xep/0313_message_archive_management.vala"

5
xmpp-vala/src/module/stanza_error.vala
View file

@ -75,7 +75,7 @@ namespace Xmpp {
error_node.put_node(new StanzaNode.build("text", ERROR_NS_URI)
.add_self_xmlns()
.put_attribute("xml:lang", "en")
.put_node(new StanzaNode.text(text))
.put_node(new StanzaNode.text(human_readable))
);
}
}
@ -94,6 +94,9 @@ namespace Xmpp {
public ErrorStanza.not_allowed(string? human_readable = null) {
this.build(TYPE_CANCEL, CONDITION_NOT_ALLOWED, human_readable, null);
}
public ErrorStanza.resource_constraint(string? human_readable = null) {
this.build(TYPE_WAIT, CONDITION_RESOURCE_CONSTRAINT, human_readable, null);
}
public ErrorStanza.service_unavailable() {
this.build(TYPE_CANCEL, CONDITION_SERVICE_UNAVAILABLE, null, null);
}

59
xmpp-vala/src/module/xep/0047_in_band_bytestreams.vala
View file

@ -14,7 +14,9 @@ public class Module : XmppStreamModule, Iq.Handler {
stream.add_flag(new Flag());
stream.get_module(Iq.Module.IDENTITY).register_for_namespace(NS_URI, this);
}
public override void detach(XmppStream stream) { }
public override void detach(XmppStream stream) {
stream.get_module(Iq.Module.IDENTITY).unregister_from_namespace(NS_URI, this);
}
public void on_iq_set(XmppStream stream, Iq.Stanza iq) {
// the iq module ensures that there's only one child node
@ -23,28 +25,28 @@ public class Module : XmppStreamModule, Iq.Handler {
node = (node != null) ? node : iq.stanza.get_subnode("data", NS_URI);
node = (node != null) ? node : iq.stanza.get_subnode("close", NS_URI);
if (node == null) {
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("unknown IBB action")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("unknown IBB action")) { to=iq.from });
return;
}
string? sid = node.get_attribute("sid");
if (sid == null) {
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("missing sid")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("missing sid")) { to=iq.from });
return;
}
Connection? conn = stream.get_flag(Flag.IDENTITY).get_connection(sid);
if (node.name == "open") {
if (conn == null) {
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.not_acceptable("unexpected IBB connection")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.not_acceptable("unexpected IBB connection")) { to=iq.from });
return;
}
if (conn.state != Connection.State.WAITING_FOR_CONNECT) {
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("IBB open for already open connection")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("IBB open for already open connection")) { to=iq.from });
return;
}
conn.handle_open(stream, node, iq);
} else {
if (conn == null || conn.state != Connection.State.CONNECTED) {
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.item_not_found()));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.item_not_found()) { to=iq.from });
return;
}
if (node.name == "close") {
@ -60,9 +62,8 @@ public class Module : XmppStreamModule, Iq.Handler {
}
public class Connection : IOStream {
// TODO(hrxi): Fix reference cycle
public class Input : InputStream {
private Connection connection;
private weak Connection connection;
public Input(Connection connection) {
this.connection = connection;
}
@ -73,14 +74,14 @@ public class Connection : IOStream {
return yield connection.read_async(buffer, io_priority, cancellable);
}
public override bool close(Cancellable? cancellable = null) throws IOError {
return connection.close_read(cancellable);
throw new IOError.NOT_SUPPORTED("can't do non-async closes on in-band bytestreams");
}
public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield connection.close_read_async(io_priority, cancellable);
}
}
public class Output : OutputStream {
private Connection connection;
private weak Connection connection;
public Output(Connection connection) {
this.connection = connection;
}
@ -91,7 +92,7 @@ public class Connection : IOStream {
return yield connection.write_async(buffer, io_priority, cancellable);
}
public override bool close(Cancellable? cancellable = null) throws IOError {
return connection.close_write(cancellable);
throw new IOError.NOT_SUPPORTED("can't do non-async closes on in-band bytestreams");
}
public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield connection.close_write_async(io_priority, cancellable);
@ -150,25 +151,25 @@ public class Connection : IOStream {
output = new Output(this);
}
public void set_read_callback(SourceFunc callback, Cancellable? cancellable, int io_priority) throws IOError {
public void set_read_callback(owned SourceFunc callback, Cancellable? cancellable, int io_priority) throws IOError {
if (read_callback != null) {
throw new IOError.PENDING("only one async read is permitted at a time on an in-band bytestream");
}
if (cancellable != null) {
read_callback_cancellable_id = cancellable.connect(trigger_read_callback);
}
read_callback = callback;
read_callback = (owned)callback;
read_callback_cancellable = cancellable;
read_callback_priority = io_priority;
}
public void set_write_callback(SourceFunc callback, Cancellable? cancellable, int io_priority) throws IOError {
public void set_write_callback(owned SourceFunc callback, Cancellable? cancellable, int io_priority) throws IOError {
if (write_callback != null) {
throw new IOError.PENDING("only one async write is permitted at a time on an in-band bytestream");
}
if (cancellable != null) {
write_callback_cancellable_id = cancellable.connect(trigger_write_callback);
}
write_callback = callback;
write_callback = (owned)callback;
write_callback_cancellable = cancellable;
write_callback_priority = io_priority;
}
@ -263,13 +264,6 @@ public class Connection : IOStream {
return buffer.length;
}
public bool close_read(Cancellable? cancellable = null) {
input_closed = true;
if (!output_closed) {
return true;
}
return close_impl(cancellable);
}
public async bool close_read_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
input_closed = true;
if (!output_closed) {
@ -277,13 +271,6 @@ public class Connection : IOStream {
}
return yield close_async_impl(io_priority, cancellable);
}
public bool close_write(Cancellable? cancellable = null) {
output_closed = true;
if (!input_closed) {
return true;
}
return close_impl(cancellable);
}
public async bool close_write_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
output_closed = true;
if (!input_closed) {
@ -292,7 +279,7 @@ public class Connection : IOStream {
return yield close_async_impl(io_priority, cancellable);
}
delegate void OnClose(bool success);
private bool close_impl(Cancellable? cancellable = null, OnClose? on_close = null) {
private bool close_impl(Cancellable? cancellable, OnClose on_close) {
if (state == State.DISCONNECTING || state == State.DISCONNECTED || state == State.ERROR) {
on_close(true);
return true;
@ -386,17 +373,17 @@ public class Connection : IOStream {
string? stanza = open.get_attribute("stanza");
if (block_size < 0 || (stanza != null && stanza != "iq" && stanza != "message")) {
set_error("invalid open");
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("missing block_size or invalid stanza")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("missing block_size or invalid stanza")) { to=iq.from });
return;
}
if (stanza != null && stanza != "iq") {
set_error("invalid open");
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.feature_not_implemented("cannot use message stanzas for IBB")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.feature_not_implemented("cannot use message stanzas for IBB")) { to=iq.from });
return;
}
if (block_size > this.block_size) {
set_error("invalid open");
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_RESOURCE_CONSTRAINT, "opening a connection with a greater than negotiated/acceptable block size", null)));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_RESOURCE_CONSTRAINT, "opening a connection with a greater than negotiated/acceptable block size", null)) { to=iq.from });
return;
}
this.block_size = block_size;
@ -408,7 +395,7 @@ public class Connection : IOStream {
assert(state == State.CONNECTED);
if (input_closed) {
set_error("unexpected data");
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.not_allowed("unexpected data")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.not_allowed("unexpected data")) { to=iq.from });
return;
}
int seq = data.get_attribute_int("seq");
@ -417,12 +404,12 @@ public class Connection : IOStream {
uint8[] content = Base64.decode(data.get_string_content());
if (content.length > block_size) {
set_error("data longer than negotiated block size");
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("data longer than negotiated block size")));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("data longer than negotiated block size")) { to=iq.from });
return;
}
if (seq < 0 || seq != remote_seq) {
set_error("out of order data packets");
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_UNEXPECTED_REQUEST, "out of order data packets", null)));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_UNEXPECTED_REQUEST, "out of order data packets", null)) { to=iq.from });
return;
}
remote_seq = (remote_seq + 1) % SEQ_MODULUS;

83
xmpp-vala/src/module/xep/0065_socks5_bytestreams.vala Normal file
View file

@ -0,0 +1,83 @@
using Gee;
using Xmpp;
using Xmpp.Xep;
namespace Xmpp.Xep.Socks5Bytestreams {
internal const string NS_URI = "http://jabber.org/protocol/bytestreams";
public class Proxy : Object {
public string host { get; private set; }
public Jid jid { get; private set; }
public int port { get; private set; }
public Proxy(string host, Jid jid, int port) {
this.host = host;
this.jid = jid;
this.port = port;
}
}
public class Module : XmppStreamModule, Iq.Handler {
public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0065_socks5_bytestreams");
public override void attach(XmppStream stream) {
stream.add_flag(new Flag());
query_availability(stream);
}
public override void detach(XmppStream stream) { }
public void on_iq_set(XmppStream stream, Iq.Stanza iq) { }
public Gee.List<Proxy> get_proxies(XmppStream stream) {
return stream.get_flag(Flag.IDENTITY).proxies;
}
private void query_availability(XmppStream stream) {
stream.get_module(ServiceDiscovery.Module.IDENTITY).request_items(stream, stream.remote_name, (stream, items_result) => {
foreach (Xep.ServiceDiscovery.Item item in items_result.items) {
stream.get_module(ServiceDiscovery.Module.IDENTITY).request_info(stream, item.jid, (stream, info_result) => {
foreach (string feature in info_result.features) {
if (feature == NS_URI) {
StanzaNode query_ = new StanzaNode.build("query", NS_URI).add_self_xmlns();
Iq.Stanza iq = new Iq.Stanza.get(query_) { to=item.jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => {
if (iq.is_error()) {
return;
}
StanzaNode? query = iq.stanza.get_subnode("query", NS_URI);
StanzaNode? stream_host = query != null ? query.get_subnode("streamhost", NS_URI) : null;
if (query == null || stream_host == null) {
return;
}
string? host = stream_host.get_attribute("host");
string? jid_str = stream_host.get_attribute("jid");
Jid? jid = jid_str != null ? Jid.parse(jid_str) : null;
int port = stream_host.get_attribute_int("port");
if (host == null || jid == null || port <= 0 || port > 65535) {
return;
}
stream.get_flag(Flag.IDENTITY).proxies.add(new Proxy(host, jid, port));
});
}
}
});
}
});
}
public override string get_ns() { return NS_URI; }
public override string get_id() { return IDENTITY.id; }
}
public class Flag : XmppStreamFlag {
public static FlagIdentity<Flag> IDENTITY = new FlagIdentity<Flag>(NS_URI, "socks5_bytestreams");
public Gee.List<Proxy> proxies = new ArrayList<Proxy>();
public override string get_ns() { return NS_URI; }
public override string get_id() { return IDENTITY.id; }
}
}

723
xmpp-vala/src/module/xep/0166_jingle.vala
View file

@ -11,7 +11,9 @@ public errordomain IqError {
BAD_REQUEST,
NOT_ACCEPTABLE,
NOT_IMPLEMENTED,
UNSUPPORTED_INFO,
OUT_OF_ORDER,
RESOURCE_CONSTRAINT,
}
void send_iq_error(IqError iq_error, XmppStream stream, Iq.Stanza iq) {
@ -22,13 +24,18 @@ void send_iq_error(IqError iq_error, XmppStream stream, Iq.Stanza iq) {
error = new ErrorStanza.not_acceptable(iq_error.message);
} else if (iq_error is IqError.NOT_IMPLEMENTED) {
error = new ErrorStanza.feature_not_implemented(iq_error.message);
} else if (iq_error is IqError.UNSUPPORTED_INFO) {
StanzaNode unsupported_info = new StanzaNode.build("unsupported-info", ERROR_NS_URI).add_self_xmlns();
error = new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_FEATURE_NOT_IMPLEMENTED, iq_error.message, unsupported_info);
} else if (iq_error is IqError.OUT_OF_ORDER) {
StanzaNode out_of_order = new StanzaNode.build("out-of-order", ERROR_NS_URI).add_self_xmlns();
error = new ErrorStanza.build(ErrorStanza.TYPE_MODIFY, ErrorStanza.CONDITION_UNEXPECTED_REQUEST, iq_error.message, out_of_order);
} else if (iq_error is IqError.RESOURCE_CONSTRAINT) {
error = new ErrorStanza.resource_constraint(iq_error.message);
} else {
assert_not_reached();
}
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, error));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, error) { to=iq.from });
}
public errordomain Error {
@ -40,34 +47,83 @@ public errordomain Error {
TRANSPORT_ERROR,
}
StanzaNode get_single_node_anyns(StanzaNode parent, string node_name) throws IqError {
StanzaNode? get_single_node_anyns(StanzaNode parent, string? node_name = null) throws IqError {
StanzaNode? result = null;
foreach (StanzaNode child in parent.get_all_subnodes()) {
if (child.name == node_name) {
if (node_name == null || child.name == node_name) {
if (result != null) {
if (node_name != null) {
throw new IqError.BAD_REQUEST(@"multiple $(node_name) nodes");
} else {
throw new IqError.BAD_REQUEST(@"expected single subnode");
}
}
result = child;
}
}
if (result == null) {
throw new IqError.BAD_REQUEST(@"missing $(node_name) node");
}
return result;
}
class ContentNode {
public Role creator;
public string name;
public StanzaNode? description;
public StanzaNode? transport;
}
ContentNode get_single_content_node(StanzaNode jingle) throws IqError {
Gee.List<StanzaNode> contents = jingle.get_subnodes("content");
if (contents.size == 0) {
throw new IqError.BAD_REQUEST("missing content node");
}
if (contents.size > 1) {
throw new IqError.NOT_IMPLEMENTED("can't process multiple content nodes");
}
StanzaNode content = contents[0];
string? creator_str = content.get_attribute("creator");
// Vala can't typecheck the ternary operator here.
Role? creator = null;
if (creator_str != null) {
creator = Role.parse(creator_str);
} else {
// TODO(hrxi): now, is the creator attribute optional or not (XEP-0166
// Jingle)?
creator = Role.INITIATOR;
}
string? name = content.get_attribute("name");
StanzaNode? description = get_single_node_anyns(content, "description");
StanzaNode? transport = get_single_node_anyns(content, "transport");
if (name == null || creator == null) {
throw new IqError.BAD_REQUEST("missing name or creator");
}
return new ContentNode() {
creator=creator,
name=name,
description=description,
transport=transport
};
}
// This module can only be attached to one stream at a time.
public class Module : XmppStreamModule, Iq.Handler {
public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0166_jingle");
private HashMap<string, ContentType> content_types = new HashMap<string, ContentType>();
private HashMap<string, Transport> transports = new HashMap<string, Transport>();
private XmppStream? current_stream = null;
public override void attach(XmppStream stream) {
stream.add_flag(new Flag());
stream.get_module(ServiceDiscovery.Module.IDENTITY).add_feature(stream, NS_URI);
stream.get_module(Iq.Module.IDENTITY).register_for_namespace(NS_URI, this);
current_stream = stream;
}
public override void detach(XmppStream stream) {
stream.get_module(Iq.Module.IDENTITY).unregister_from_namespace(NS_URI, this);
}
public override void detach(XmppStream stream) { }
public void register_content_type(ContentType content_type) {
content_types[content_type.content_type_ns_uri()] = content_type;
@ -87,17 +143,25 @@ public class Module : XmppStreamModule, Iq.Handler {
}
return transports[ns_uri];
}
public Transport? select_transport(XmppStream stream, TransportType type, Jid receiver_full_jid) {
public Transport? select_transport(XmppStream stream, TransportType type, Jid receiver_full_jid, Set<string> blacklist) {
Transport? result = null;
foreach (Transport transport in transports.values) {
if (transport.transport_type() != type) {
continue;
}
// TODO(hrxi): prioritization
if (transport.transport_ns_uri() in blacklist) {
continue;
}
if (transport.is_transport_available(stream, receiver_full_jid)) {
return transport;
if (result != null) {
if (result.transport_priority() >= transport.transport_priority()) {
continue;
}
}
return null;
result = transport;
}
}
return result;
}
private bool is_jingle_available(XmppStream stream, Jid full_jid) {
@ -106,14 +170,14 @@ public class Module : XmppStreamModule, Iq.Handler {
}
public bool is_available(XmppStream stream, TransportType type, Jid full_jid) {
return is_jingle_available(stream, full_jid) && select_transport(stream, type, full_jid) != null;
return is_jingle_available(stream, full_jid) && select_transport(stream, type, full_jid, Set.empty()) != null;
}
public Session create_session(XmppStream stream, TransportType type, Jid receiver_full_jid, Senders senders, string content_name, StanzaNode description) throws Error {
if (!is_jingle_available(stream, receiver_full_jid)) {
throw new Error.NO_SHARED_PROTOCOLS("No Jingle support");
}
Transport? transport = select_transport(stream, type, receiver_full_jid);
Transport? transport = select_transport(stream, type, receiver_full_jid, Set.empty());
if (transport == null) {
throw new Error.NO_SHARED_PROTOCOLS("No suitable transports");
}
@ -121,8 +185,8 @@ public class Module : XmppStreamModule, Iq.Handler {
if (my_jid == null) {
throw new Error.GENERAL("Couldn't determine own JID");
}
TransportParameters transport_params = transport.create_transport_parameters();
Session session = new Session.initiate_sent(random_uuid(), type, transport_params, receiver_full_jid, content_name);
TransportParameters transport_params = transport.create_transport_parameters(stream, my_jid, receiver_full_jid);
Session session = new Session.initiate_sent(random_uuid(), type, transport_params, my_jid, receiver_full_jid, content_name, send_terminate_and_remove_session);
StanzaNode content = new StanzaNode.build("content", NS_URI)
.put_attribute("creator", "initiator")
.put_attribute("name", content_name)
@ -146,51 +210,58 @@ public class Module : XmppStreamModule, Iq.Handler {
}
public void handle_session_initiate(XmppStream stream, string sid, StanzaNode jingle, Iq.Stanza iq) throws IqError {
Gee.List<StanzaNode> contents = jingle.get_subnodes("content");
if (contents.size == 0) {
throw new IqError.BAD_REQUEST("missing content node");
ContentNode content = get_single_content_node(jingle);
if (content.description == null || content.transport == null) {
throw new IqError.BAD_REQUEST("missing description or transport node");
}
if (contents.size > 1) {
throw new IqError.NOT_IMPLEMENTED("can't process multiple content nodes");
Jid? my_jid = stream.get_flag(Bind.Flag.IDENTITY).my_jid;
if (my_jid == null) {
throw new IqError.RESOURCE_CONSTRAINT("Couldn't determine own JID");
}
StanzaNode content = contents[0];
string? name = content.get_attribute("name");
StanzaNode description = get_single_node_anyns(content, "description");
StanzaNode transport_node = get_single_node_anyns(content, "transport");
if (name == null) {
throw new IqError.BAD_REQUEST("missing name");
}
Transport? transport = get_transport(transport_node.ns_uri);
Transport? transport = get_transport(content.transport.ns_uri);
TransportParameters? transport_params = null;
if (transport != null) {
transport_params = transport.parse_transport_parameters(transport_node);
transport_params = transport.parse_transport_parameters(stream, my_jid, iq.from, content.transport);
} else {
// terminate the session below
}
ContentType? content_type = get_content_type(description.ns_uri);
ContentType? content_type = get_content_type(content.description.ns_uri);
if (content_type == null) {
// TODO(hrxi): how do we signal an unknown content type?
throw new IqError.NOT_IMPLEMENTED("unknown content type");
}
ContentParameters content_params = content_type.parse_content_parameters(description);
ContentParameters content_params = content_type.parse_content_parameters(content.description);
TransportType type = content_type.content_type_transport_type();
Session session = new Session.initiate_received(sid, type, transport_params, iq.from, name);
Session session = new Session.initiate_received(sid, type, transport_params, my_jid, iq.from, content.name, send_terminate_and_remove_session);
stream.get_flag(Flag.IDENTITY).add_session(session);
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
if (transport == null || transport.transport_type() != type) {
StanzaNode reason = new StanzaNode.build("reason", NS_URI)
.put_node(new StanzaNode.build("unsupported-transports", NS_URI));
session.terminate(stream, reason);
session.terminate(reason, "unsupported transports");
return;
}
content_params.on_session_initiate(stream, session);
}
private void send_terminate_and_remove_session(Jid to, string sid, StanzaNode reason) {
StanzaNode jingle = new StanzaNode.build("jingle", NS_URI)
.add_self_xmlns()
.put_attribute("action", "session-terminate")
.put_attribute("sid", sid)
.put_node(reason);
Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=to };
current_stream.get_module(Iq.Module.IDENTITY).send_iq(current_stream, iq);
// Immediately remove the session from the open sessions as per the
// XEP, don't wait for confirmation.
current_stream.get_flag(Flag.IDENTITY).remove_session(sid);
}
public void on_iq_set(XmppStream stream, Iq.Stanza iq) {
try {
handle_iq_set(stream, iq);
@ -210,7 +281,7 @@ public class Module : XmppStreamModule, Iq.Handler {
if (action == "session-initiate") {
if (session != null) {
// TODO(hrxi): Info leak if other clients use predictable session IDs?
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_MODIFY, ErrorStanza.CONDITION_CONFLICT, "session ID already in use", null)));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_MODIFY, ErrorStanza.CONDITION_CONFLICT, "session ID already in use", null)) { to=iq.from });
return;
}
handle_session_initiate(stream, sid, jingle, iq);
@ -218,7 +289,7 @@ public class Module : XmppStreamModule, Iq.Handler {
}
if (session == null) {
StanzaNode unknown_session = new StanzaNode.build("unknown-session", ERROR_NS_URI).add_self_xmlns();
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.item_not_found(unknown_session)));
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.item_not_found(unknown_session)) { to=iq.from });
return;
}
session.handle_iq_set(stream, action, jingle, iq);
@ -250,19 +321,26 @@ public enum Senders {
}
}
public delegate void SessionTerminate(Jid to, string sid, StanzaNode reason);
public interface Transport : Object {
public abstract string transport_ns_uri();
public abstract bool is_transport_available(XmppStream stream, Jid full_jid);
public abstract TransportType transport_type();
public abstract TransportParameters create_transport_parameters();
public abstract TransportParameters parse_transport_parameters(StanzaNode transport) throws IqError;
public abstract int transport_priority();
public abstract TransportParameters create_transport_parameters(XmppStream stream, Jid local_full_jid, Jid peer_full_jid);
public abstract TransportParameters parse_transport_parameters(XmppStream stream, Jid local_full_jid, Jid peer_full_jid, StanzaNode transport) throws IqError;
}
// Gets a null `stream` if connection setup was unsuccessful and another
// transport method should be tried.
public interface TransportParameters : Object {
public abstract string transport_ns_uri();
public abstract StanzaNode to_transport_stanza_node();
public abstract void update_transport(StanzaNode transport) throws IqError;
public abstract IOStream create_transport_connection(XmppStream stream, Jid peer_full_jid, Role role);
public abstract void on_transport_accept(StanzaNode transport) throws IqError;
public abstract void on_transport_info(StanzaNode transport) throws IqError;
public abstract void create_transport_connection(XmppStream stream, Session session);
}
public enum Role {
@ -276,12 +354,21 @@ public enum Role {
}
assert_not_reached();
}
public static Role parse(string role) throws IqError {
switch (role) {
case "initiator": return INITIATOR;
case "responder": return RESPONDER;
}
throw new IqError.BAD_REQUEST(@"invalid role $(role)");
}
}
public interface ContentType : Object {
public abstract string content_type_ns_uri();
public abstract TransportType content_type_transport_type();
public abstract ContentParameters parse_content_parameters(StanzaNode description) throws IqError;
public abstract void handle_content_session_info(XmppStream stream, Session session, StanzaNode info, Iq.Stanza iq) throws IqError;
}
public interface ContentParameters : Object {
@ -290,79 +377,154 @@ public interface ContentParameters : Object {
public class Session {
// INITIATE_SENT -> ACTIVE -> ENDED
// INITIATE_RECEIVED -> ACTIVE -> ENDED
// INITIATE_SENT -> CONNECTING -> [REPLACING_TRANSPORT -> CONNECTING ->]... ACTIVE -> ENDED
// INITIATE_RECEIVED -> CONNECTING -> [WAITING_FOR_TRANSPORT_REPLACE -> CONNECTING ->].. ACTIVE -> ENDED
public enum State {
INITIATE_SENT,
REPLACING_TRANSPORT,
INITIATE_RECEIVED,
WAITING_FOR_TRANSPORT_REPLACE,
CONNECTING,
ACTIVE,
ENDED,
}
public State state { get; private set; }
public Role role { get; private set; }
public string sid { get; private set; }
public Type type_ { get; private set; }
public TransportType type_ { get; private set; }
public Jid local_full_jid { get; private set; }
public Jid peer_full_jid { get; private set; }
public Role content_creator { get; private set; }
public string content_name { get; private set; }
// INITIATE_SENT | INITIATE_RECEIVED
private Connection connection;
public IOStream conn { get { return connection; } }
public bool terminate_on_connection_close { get; set; }
// INITIATE_SENT | INITIATE_RECEIVED | CONNECTING
Set<string> tried_transport_methods = new HashSet<string>();
TransportParameters? transport = null;
// ACTIVE
public IOStream? conn { get; private set; }
SessionTerminate session_terminate_handler;
// Only interesting in INITIATE_SENT.
// Signals that the session has been accepted by the peer.
public signal void accepted(XmppStream stream);
public Session.initiate_sent(string sid, Type type, TransportParameters transport, Jid peer_full_jid, string content_name) {
public Session.initiate_sent(string sid, TransportType type, TransportParameters transport, Jid local_full_jid, Jid peer_full_jid, string content_name, owned SessionTerminate session_terminate_handler) {
this.state = State.INITIATE_SENT;
this.role = Role.INITIATOR;
this.sid = sid;
this.type_ = type;
this.local_full_jid = local_full_jid;
this.peer_full_jid = peer_full_jid;
this.content_creator = Role.INITIATOR;
this.content_name = content_name;
this.tried_transport_methods = new HashSet<string>();
this.tried_transport_methods.add(transport.transport_ns_uri());
this.transport = transport;
this.conn = null;
this.connection = new Connection(this);
this.session_terminate_handler = (owned)session_terminate_handler;
this.terminate_on_connection_close = true;
}
public Session.initiate_received(string sid, Type type, TransportParameters? transport, Jid peer_full_jid, string content_name) {
public Session.initiate_received(string sid, TransportType type, TransportParameters? transport, Jid local_full_jid, Jid peer_full_jid, string content_name, owned SessionTerminate session_terminate_handler) {
this.state = State.INITIATE_RECEIVED;
this.role = Role.RESPONDER;
this.sid = sid;
this.type_ = type;
this.local_full_jid = local_full_jid;
this.peer_full_jid = peer_full_jid;
this.content_creator = Role.INITIATOR;
this.content_name = content_name;
this.transport = transport;
this.conn = null;
this.tried_transport_methods = new HashSet<string>();
if (transport != null) {
this.tried_transport_methods.add(transport.transport_ns_uri());
}
this.connection = new Connection(this);
this.session_terminate_handler = (owned)session_terminate_handler;
this.terminate_on_connection_close = true;
}
public void handle_iq_set(XmppStream stream, string action, StanzaNode jingle, Iq.Stanza iq) throws IqError {
// Validate action.
switch (action) {
case "session-accept":
if (state != State.INITIATE_SENT) {
throw new IqError.OUT_OF_ORDER("got session-accept while not waiting for one");
}
handle_session_accept(stream, jingle, iq);
break;
case "session-info":
case "session-terminate":
handle_session_terminate(stream, jingle, iq);
case "transport-accept":
case "transport-info":
case "transport-reject":
case "transport-replace":
break;
case "content-accept":
case "content-add":
case "content-modify":
case "content-reject":
case "content-remove":
case "description-info":
case "security-info":
case "transport-accept":
case "transport-info":
case "transport-reject":
case "transport-replace":
throw new IqError.NOT_IMPLEMENTED(@"$(action) is not implemented");
default:
throw new IqError.BAD_REQUEST("invalid action");
}
ContentNode? content = null;
StanzaNode? transport = null;
// Do some pre-processing.
if (action != "session-info" && action != "session-terminate") {
content = get_single_content_node(jingle);
verify_content(content);
switch (action) {
case "transport-accept":
case "transport-reject":
case "transport-replace":
case "transport-info":
switch (state) {
case State.INITIATE_SENT:
case State.REPLACING_TRANSPORT:
case State.INITIATE_RECEIVED:
case State.WAITING_FOR_TRANSPORT_REPLACE:
case State.CONNECTING:
break;
default:
throw new IqError.OUT_OF_ORDER("transport-* unsupported after connection setup");
}
void handle_session_accept(XmppStream stream, StanzaNode jingle, Iq.Stanza iq) throws IqError {
// TODO(hrxi): What to do with description nodes?
if (content.transport == null) {
throw new IqError.BAD_REQUEST("missing transport node");
}
transport = content.transport;
break;
}
}
switch (action) {
case "session-accept":
if (state != State.INITIATE_SENT) {
throw new IqError.OUT_OF_ORDER("got session-accept while not waiting for one");
}
handle_session_accept(stream, content, jingle, iq);
break;
case "session-info":
handle_session_info(stream, jingle, iq);
break;
case "session-terminate":
handle_session_terminate(stream, jingle, iq);
break;
case "transport-accept":
handle_transport_accept(stream, transport, jingle, iq);
break;
case "transport-reject":
handle_transport_reject(stream, jingle, iq);
break;
case "transport-replace":
handle_transport_replace(stream, transport, jingle, iq);
break;
case "transport-info":
handle_transport_info(stream, transport, jingle, iq);
break;
}
}
void handle_session_accept(XmppStream stream, ContentNode content, StanzaNode jingle, Iq.Stanza iq) throws IqError {
string? responder_str = jingle.get_attribute("responder");
Jid responder;
if (responder_str != null) {
@ -374,32 +536,168 @@ public class Session {
if (!responder.is_full()) {
throw new IqError.BAD_REQUEST("invalid responder JID");
}
Gee.List<StanzaNode> contents = jingle.get_subnodes("content");
if (contents.size == 0) {
// TODO(hrxi): here and below, should we terminate the session?
throw new IqError.BAD_REQUEST("missing content node");
if (content.description == null || content.transport == null) {
throw new IqError.BAD_REQUEST("missing description or transport node");
}
if (contents.size > 1) {
throw new IqError.NOT_IMPLEMENTED("can't process multiple content nodes");
}
StanzaNode content = contents[0];
StanzaNode description = get_single_node_anyns(content, "description");
StanzaNode transport_node = get_single_node_anyns(content, "transport");
if (transport_node.ns_uri != transport.transport_ns_uri()) {
if (content.transport.ns_uri != transport.transport_ns_uri()) {
throw new IqError.BAD_REQUEST("session-accept with unnegotiated transport method");
}
transport.update_transport(transport_node);
conn = transport.create_transport_connection(stream, peer_full_jid, Role.INITIATOR);
transport = null;
transport.on_transport_accept(content.transport);
StanzaNode description = content.description; // TODO(hrxi): handle this :P
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
state = State.CONNECTING;
transport.create_transport_connection(stream, this);
}
void connection_created(XmppStream stream, IOStream? conn) {
if (state != State.CONNECTING) {
return;
}
if (conn != null) {
state = State.ACTIVE;
accepted(stream);
transport = null;
tried_transport_methods.clear();
connection.set_inner(conn);
} else {
if (role == Role.INITIATOR) {
select_new_transport(stream);
} else {
state = State.WAITING_FOR_TRANSPORT_REPLACE;
}
}
}
void handle_session_terminate(XmppStream stream, StanzaNode jingle, Iq.Stanza iq) throws IqError {
connection.on_terminated_by_jingle("remote terminated jingle session");
state = State.ENDED;
stream.get_flag(Flag.IDENTITY).remove_session(sid);
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
// TODO(hrxi): also handle presence type=unavailable
}
void handle_session_info(XmppStream stream, StanzaNode jingle, Iq.Stanza iq) throws IqError {
StanzaNode? info = get_single_node_anyns(jingle);
if (info == null) {
// Jingle session ping
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
return;
}
ContentType? content_type = stream.get_module(Module.IDENTITY).get_content_type(info.ns_uri);
if (content_type == null) {
throw new IqError.UNSUPPORTED_INFO("unknown session-info namespace");
}
content_type.handle_content_session_info(stream, this, info, iq);
}
void select_new_transport(XmppStream stream) {
Transport? new_transport = stream.get_module(Module.IDENTITY).select_transport(stream, type_, peer_full_jid, tried_transport_methods);
if (new_transport == null) {
StanzaNode reason = new StanzaNode.build("reason", NS_URI)
.put_node(new StanzaNode.build("failed-transport", NS_URI));
terminate(reason, "failed transport");
return;
}
tried_transport_methods.add(new_transport.transport_ns_uri());
transport = new_transport.create_transport_parameters(stream, local_full_jid, peer_full_jid);
StanzaNode jingle = new StanzaNode.build("jingle", NS_URI)
.add_self_xmlns()
.put_attribute("action", "transport-replace")
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("content", NS_URI)
.put_attribute("creator", "initiator")
.put_attribute("name", content_name)
.put_node(transport.to_transport_stanza_node())
);
Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=peer_full_jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq);
state = State.REPLACING_TRANSPORT;
}
void handle_transport_accept(XmppStream stream, StanzaNode transport_node, StanzaNode jingle, Iq.Stanza iq) throws IqError {
if (state != State.REPLACING_TRANSPORT) {
throw new IqError.OUT_OF_ORDER("no outstanding transport-replace request");
}
if (transport_node.ns_uri != transport.transport_ns_uri()) {
throw new IqError.BAD_REQUEST("transport-accept with unnegotiated transport method");
}
transport.on_transport_accept(transport_node);
state = State.CONNECTING;
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
transport.create_transport_connection(stream, this);
}
void handle_transport_reject(XmppStream stream, StanzaNode jingle, Iq.Stanza iq) throws IqError {
if (state != State.REPLACING_TRANSPORT) {
throw new IqError.OUT_OF_ORDER("no outstanding transport-replace request");
}
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
select_new_transport(stream);
}
void handle_transport_replace(XmppStream stream, StanzaNode transport_node, StanzaNode jingle, Iq.Stanza iq) throws IqError {
Transport? transport = stream.get_module(Module.IDENTITY).get_transport(transport_node.ns_uri);
TransportParameters? parameters = null;
if (transport != null) {
// Just parse the transport info for the errors.
parameters = transport.parse_transport_parameters(stream, local_full_jid, peer_full_jid, transport_node);
}
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
if (state != State.WAITING_FOR_TRANSPORT_REPLACE || transport == null) {
StanzaNode jingle_response = new StanzaNode.build("jingle", NS_URI)
.add_self_xmlns()
.put_attribute("action", "transport-reject")
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("content", NS_URI)
.put_attribute("creator", "initiator")
.put_attribute("name", content_name)
.put_node(transport_node)
);
Iq.Stanza iq_response = new Iq.Stanza.set(jingle_response) { to=peer_full_jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq_response);
return;
}
this.transport = parameters;
StanzaNode jingle_response = new StanzaNode.build("jingle", NS_URI)
.add_self_xmlns()
.put_attribute("action", "transport-accept")
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("content", NS_URI)
.put_attribute("creator", "initiator")
.put_attribute("name", content_name)
.put_node(this.transport.to_transport_stanza_node())
);
Iq.Stanza iq_response = new Iq.Stanza.set(jingle_response) { to=peer_full_jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq_response);
state = State.CONNECTING;
this.transport.create_transport_connection(stream, this);
}
void handle_transport_info(XmppStream stream, StanzaNode transport, StanzaNode jingle, Iq.Stanza iq) throws IqError {
this.transport.on_transport_info(transport);
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
}
void verify_content(ContentNode content) throws IqError {
if (content.name != content_name || content.creator != content_creator) {
throw new IqError.BAD_REQUEST("unknown content");
}
}
public void set_transport_connection(XmppStream stream, IOStream? conn) {
if (state != State.CONNECTING) {
return;
}
connection_created(stream, conn);
}
public void send_transport_info(XmppStream stream, StanzaNode transport) {
if (state != State.CONNECTING) {
return;
}
StanzaNode jingle = new StanzaNode.build("jingle", NS_URI)
.add_self_xmlns()
.put_attribute("action", "transport-info")
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("content", NS_URI)
.put_attribute("creator", "initiator")
.put_attribute("name", content_name)
.put_node(transport)
);
Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=peer_full_jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq);
}
public void accept(XmppStream stream, StanzaNode description) {
if (state != State.INITIATE_RECEIVED) {
return; // TODO(hrxi): what to do?
@ -417,10 +715,8 @@ public class Session {
Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=peer_full_jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq);
conn = transport.create_transport_connection(stream, peer_full_jid, Role.RESPONDER);
transport = null;
state = State.ACTIVE;
state = State.CONNECTING;
transport.create_transport_connection(stream, this);
}
public void reject(XmppStream stream) {
@ -429,7 +725,7 @@ public class Session {
}
StanzaNode reason = new StanzaNode.build("reason", NS_URI)
.put_node(new StanzaNode.build("decline", NS_URI));
terminate(stream, reason);
terminate(reason, "declined");
}
public void set_application_error(XmppStream stream, StanzaNode? application_reason = null) {
@ -438,37 +734,254 @@ public class Session {
if (application_reason != null) {
reason.put_node(application_reason);
}
terminate(stream, reason);
terminate(reason, "application error");
}
public void close_connection(XmppStream stream) {
if (state != State.ACTIVE) {
return; // TODO(hrxi): what to do?
public void on_connection_error(IOError error) {
// TODO(hrxi): where can we get an XmppStream from?
StanzaNode reason = new StanzaNode.build("reason", NS_URI)
.put_node(new StanzaNode.build("failed-transport", NS_URI))
.put_node(new StanzaNode.build("text", NS_URI)
.put_node(new StanzaNode.text(error.message))
);
terminate(reason, @"transport error: $(error.message)");
}
public void on_connection_close() {
if (terminate_on_connection_close) {
StanzaNode reason = new StanzaNode.build("reason", NS_URI)
.put_node(new StanzaNode.build("success", NS_URI));
terminate(reason, "success");
}
conn.close();
}
public void terminate(XmppStream stream, StanzaNode reason) {
if (state != State.INITIATE_SENT && state != State.INITIATE_RECEIVED && state != State.ACTIVE) {
// TODO(hrxi): what to do?
public void terminate(StanzaNode reason, string? local_reason) {
if (state == State.ENDED) {
return;
}
if (state == State.ACTIVE) {
conn.close();
if (local_reason != null) {
connection.on_terminated_by_jingle(@"local session-terminate: $(local_reason)");
} else {
connection.on_terminated_by_jingle("local session-terminate");
}
}
StanzaNode jingle = new StanzaNode.build("jingle", NS_URI)
.add_self_xmlns()
.put_attribute("action", "session-terminate")
.put_attribute("sid", sid)
.put_node(reason);
Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=peer_full_jid };
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq);
session_terminate_handler(peer_full_jid, sid, reason);
state = State.ENDED;
// Immediately remove the session from the open sessions as per the
// XEP, don't wait for confirmation.
stream.get_flag(Flag.IDENTITY).remove_session(sid);
}
}
public class Connection : IOStream {
public class Input : InputStream {
private weak Connection connection;
public Input(Connection connection) {
this.connection = connection;
}
public override ssize_t read(uint8[] buffer, Cancellable? cancellable = null) throws IOError {
throw new IOError.NOT_SUPPORTED("can't do non-async reads on jingle connections");
}
public override async ssize_t read_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield connection.read_async(buffer, io_priority, cancellable);
}
public override bool close(Cancellable? cancellable = null) throws IOError {
return connection.close_read(cancellable);
}
public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield connection.close_read_async(io_priority, cancellable);
}
}
public class Output : OutputStream {
private weak Connection connection;
public Output(Connection connection) {
this.connection = connection;
}
public override ssize_t write(uint8[] buffer, Cancellable? cancellable = null) throws IOError {
throw new IOError.NOT_SUPPORTED("can't do non-async writes on jingle connections");
}
public override async ssize_t write_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield connection.write_async(buffer, io_priority, cancellable);
}
public override bool close(Cancellable? cancellable = null) throws IOError {
return connection.close_write(cancellable);
}
public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield connection.close_write_async(io_priority, cancellable);
}
}
private Input input;
private Output output;
public override InputStream input_stream { get { return input; } }
public override OutputStream output_stream { get { return output; } }
private weak Session session;
private IOStream? inner = null;
private string? error = null;
private bool read_closed = false;
private bool write_closed = false;
private class OnSetInnerCallback {
public SourceFunc callback;
public int io_priority;
}
Gee.List<OnSetInnerCallback> callbacks = new ArrayList<OnSetInnerCallback>();
public Connection(Session session) {
this.input = new Input(this);
this.output = new Output(this);
this.session = session;
}
public void set_inner(IOStream inner) {
assert(this.inner == null);
this.inner = inner;
foreach (OnSetInnerCallback c in callbacks) {
Idle.add((owned) c.callback, c.io_priority);
}
callbacks = null;
}
public void on_terminated_by_jingle(string reason) {
if (error == null) {
close_async.begin();
error = reason;
}
}
private void check_for_errors() throws IOError {
if (error != null) {
throw new IOError.CLOSED(error);
}
}
private async void wait_and_check_for_errors(int io_priority, Cancellable? cancellable = null) throws IOError {
while (true) {
check_for_errors();
if (inner != null) {
return;
}
SourceFunc callback = wait_and_check_for_errors.callback;
ulong id = 0;
if (cancellable != null) {
id = cancellable.connect(() => callback());
}
callbacks.add(new OnSetInnerCallback() { callback=(owned)callback, io_priority=io_priority});
yield;
if (cancellable != null) {
cancellable.disconnect(id);
}
}
}
private void handle_connection_error(IOError error) {
Session? strong = session;
if (strong != null) {
strong.on_connection_error(error);
}
}
private void handle_connection_close() {
Session? strong = session;
if (strong != null) {
strong.on_connection_close();
}
}
public async ssize_t read_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
yield wait_and_check_for_errors(io_priority, cancellable);
try {
return yield inner.input_stream.read_async(buffer, io_priority, cancellable);
} catch (IOError e) {
handle_connection_error(e);
throw e;
}
}
public async ssize_t write_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
yield wait_and_check_for_errors(io_priority, cancellable);
try {
return yield inner.output_stream.write_async(buffer, io_priority, cancellable);
} catch (IOError e) {
handle_connection_error(e);
throw e;
}
}
public bool close_read(Cancellable? cancellable = null) throws IOError {
check_for_errors();
if (read_closed) {
return true;
}
close_read_async.begin(GLib.Priority.DEFAULT, cancellable);
return true;
}
public async bool close_read_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
yield wait_and_check_for_errors(io_priority, cancellable);
if (read_closed) {
return true;
}
read_closed = true;
IOError error = null;
bool result = true;
try {
result = yield inner.input_stream.close_async(io_priority, cancellable);
} catch (IOError e) {
if (error == null) {
error = e;
}
}
try {
result = (yield close_if_both_closed(io_priority, cancellable)) && result;
} catch (IOError e) {
if (error == null) {
error = e;
}
}
if (error != null) {
handle_connection_error(error);
throw error;
}
return result;
}
public bool close_write(Cancellable? cancellable = null) throws IOError {
check_for_errors();
if (write_closed) {
return true;
}
close_write_async.begin(GLib.Priority.DEFAULT, cancellable);
return true;
}
public async bool close_write_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
yield wait_and_check_for_errors(io_priority, cancellable);
if (write_closed) {
return true;
}
write_closed = true;
IOError error = null;
bool result = true;
try {
result = yield inner.output_stream.close_async(io_priority, cancellable);
} catch (IOError e) {
if (error == null) {
error = e;
}
}
try {
result = (yield close_if_both_closed(io_priority, cancellable)) && result;
} catch (IOError e) {
if (error == null) {
error = e;
}
}
if (error != null) {
handle_connection_error(error);
throw error;
}
return result;
}
private async bool close_if_both_closed(int io_priority, Cancellable? cancellable = null) throws IOError {
if (read_closed && write_closed) {
handle_connection_close();
//return yield inner.close_async(io_priority, cancellable);
}
return true;
}
}

80
xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala
View file

@ -24,6 +24,19 @@ public class Module : Jingle.ContentType, XmppStreamModule {
public Jingle.ContentParameters parse_content_parameters(StanzaNode description) throws Jingle.IqError {
return Parameters.parse(this, description);
}
public void handle_content_session_info(XmppStream stream, Jingle.Session session, StanzaNode info, Iq.Stanza iq) throws Jingle.IqError {
switch (info.name) {
case "received":
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
break;
case "checksum":
// TODO(hrxi): handle hash
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq));
break;
default:
throw new Jingle.IqError.UNSUPPORTED_INFO(@"unsupported file transfer info $(info.name)");
}
}
public signal void file_incoming(XmppStream stream, FileTransfer file_transfer);
@ -45,13 +58,9 @@ public class Module : Jingle.ContentType, XmppStreamModule {
Jingle.Session session = stream.get_module(Jingle.Module.IDENTITY)
.create_session(stream, Jingle.TransportType.STREAMING, receiver_full_jid, Jingle.Senders.INITIATOR, "a-file-offer", description); // TODO(hrxi): Why "a-file-offer"?
session.terminate_on_connection_close = false;
SourceFunc callback = offer_file_stream.callback;
session.accepted.connect((stream) => {
session.conn.input_stream.close();
Idle.add((owned) callback);
});
yield;
yield session.conn.input_stream.close_async();
// TODO(hrxi): catch errors
yield session.conn.output_stream.splice_async(input_stream, OutputStreamSpliceFlags.CLOSE_SOURCE|OutputStreamSpliceFlags.CLOSE_TARGET);
@ -69,7 +78,7 @@ public class Parameters : Jingle.ContentParameters, Object {
public int64 size { get; private set; }
public StanzaNode original_description { get; private set; }
public Parameters(Module parent, StanzaNode original_description, string? media_type, string? name, int64? size) {
public Parameters(Module parent, StanzaNode original_description, string? media_type, string? name, int64 size) {
this.parent = parent;
this.original_description = original_description;
this.media_type = media_type;
@ -91,13 +100,16 @@ public class Parameters : Jingle.ContentParameters, Object {
string? size_raw = size_node != null ? size_node.get_string_content() : null;
// TODO(hrxi): For some reason, the ?:-expression does not work due to a type error.
//int64? size = size_raw != null ? int64.parse(size_raw) : null; // TODO(hrxi): this has no error handling
int64 size = -1;
if (size_raw != null) {
size = int64.parse(size_raw);
if (size_raw == null) {
// Jingle file transfers (XEP-0234) theoretically SHOULD send a
// file size, however, we do require it in order to reliably find
// the end of the file transfer.
throw new Jingle.IqError.BAD_REQUEST("file offer without file size");
}
int64 size = int64.parse(size_raw);
if (size < 0) {
throw new Jingle.IqError.BAD_REQUEST("negative file size is invalid");
}
}
return new Parameters(parent, description, media_type, name, size);
}
@ -107,6 +119,47 @@ public class Parameters : Jingle.ContentParameters, Object {
}
}
// Does nothing except wrapping an input stream to signal EOF after reading
// `max_size` bytes.
private class FileTransferInputStream : InputStream {
InputStream inner;
int64 remaining_size;
public FileTransferInputStream(InputStream inner, int64 max_size) {
this.inner = inner;
this.remaining_size = max_size;
}
private ssize_t update_remaining(ssize_t read) {
this.remaining_size -= read;
return read;
}
public override ssize_t read(uint8[] buffer_, Cancellable? cancellable = null) throws IOError {
unowned uint8[] buffer = buffer_;
if (remaining_size <= 0) {
return 0;
}
if (buffer.length > remaining_size) {
buffer = buffer[0:remaining_size];
}
return update_remaining(inner.read(buffer, cancellable));
}
public override async ssize_t read_async(uint8[]? buffer_, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
unowned uint8[] buffer = buffer_;
if (remaining_size <= 0) {
return 0;
}
if (buffer.length > remaining_size) {
buffer = buffer[0:remaining_size];
}
return update_remaining(yield inner.read_async(buffer, io_priority, cancellable));
}
public override bool close(Cancellable? cancellable = null) throws IOError {
return inner.close(cancellable);
}
public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
return yield inner.close_async(io_priority, cancellable);
}
}
public class FileTransfer : Object {
Jingle.Session session;
Parameters parameters;
@ -115,14 +168,15 @@ public class FileTransfer : Object {
public string? file_name { get { return parameters.name; } }
public int64 size { get { return parameters.size; } }
public InputStream? stream { get { return session.conn != null ? session.conn.input_stream : null; } }
public InputStream? stream { get; private set; }
public FileTransfer(Jingle.Session session, Parameters parameters) {
this.session = session;
this.parameters = parameters;
this.stream = new FileTransferInputStream(session.conn.input_stream, parameters.size);
}
public void accept(XmppStream stream) {
public void accept(XmppStream stream) throws IOError {
session.accept(stream, parameters.original_description);
session.conn.output_stream.close();
}

533
xmpp-vala/src/module/xep/0260_jingle_socks5_bytestreams.vala Normal file
View file

@ -0,0 +1,533 @@
using Gee;
using Xmpp;
using Xmpp.Xep;
namespace Xmpp.Xep.JingleSocks5Bytestreams {
private const string NS_URI = "urn:xmpp:jingle:transports:s5b:1";
public class Module : Jingle.Transport, XmppStreamModule {
public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0260_jingle_socks5_bytestreams");
public override void attach(XmppStream stream) {
stream.get_module(Jingle.Module.IDENTITY).register_transport(this);
stream.get_module(ServiceDiscovery.Module.IDENTITY).add_feature(stream, NS_URI);
}
public override void detach(XmppStream stream) { }
public override string get_ns() { return NS_URI; }
public override string get_id() { return IDENTITY.id; }
public bool is_transport_available(XmppStream stream, Jid full_jid) {
bool? result = stream.get_flag(ServiceDiscovery.Flag.IDENTITY).has_entity_feature(full_jid, NS_URI);
return result != null && result;
}
public string transport_ns_uri() {
return NS_URI;
}
public Jingle.TransportType transport_type() {
return Jingle.TransportType.STREAMING;
}
public int transport_priority() {
return 1;
}
private Gee.List<Candidate> get_local_candidates(XmppStream stream) {
Gee.List<Candidate> result = new ArrayList<Candidate>();
int i = 1 << 15;
foreach (Socks5Bytestreams.Proxy proxy in stream.get_module(Socks5Bytestreams.Module.IDENTITY).get_proxies(stream)) {
result.add(new Candidate.proxy(random_uuid(), proxy, i));
i -= 1;
}
return result;
}
public Jingle.TransportParameters create_transport_parameters(XmppStream stream, Jid local_full_jid, Jid peer_full_jid) {
Parameters result = new Parameters.create(local_full_jid, peer_full_jid, random_uuid());
result.local_candidates.add_all(get_local_candidates(stream));
return result;
}
public Jingle.TransportParameters parse_transport_parameters(XmppStream stream, Jid local_full_jid, Jid peer_full_jid, StanzaNode transport) throws Jingle.IqError {
Parameters result = Parameters.parse(local_full_jid, peer_full_jid, transport);
result.local_candidates.add_all(get_local_candidates(stream));
return result;
}
}
public enum CandidateType {
ASSISTED,
DIRECT,
PROXY,
TUNNEL;
public static CandidateType parse(string type) throws Jingle.IqError {
switch (type) {
case "assisted": return CandidateType.ASSISTED;
case "direct": return CandidateType.DIRECT;
case "proxy": return CandidateType.PROXY;
case "tunnel": return CandidateType.TUNNEL;
}
throw new Jingle.IqError.BAD_REQUEST(@"unknown candidate type $(type)");
}
public string to_string() {
switch (this) {
case ASSISTED: return "assisted";
case DIRECT: return "direct";
case PROXY: return "proxy";
case TUNNEL: return "tunnel";
}
assert_not_reached();
}
private int type_preference_impl() {
switch (this) {
case ASSISTED: return 120;
case DIRECT: return 126;
case PROXY: return 10;
case TUNNEL: return 110;
}
assert_not_reached();
}
public int type_preference() {
return type_preference_impl() << 16;
}
}
public class Candidate : Socks5Bytestreams.Proxy {
public string cid { get; private set; }
public int priority { get; private set; }
public CandidateType type_ { get; private set; }
private Candidate(string cid, string host, Jid jid, int port, int priority, CandidateType type) {
base(host, jid, port);
this.cid = cid;
this.priority = priority;
this.type_ = type;
}
public Candidate.build(string cid, string host, Jid jid, int port, int local_priority, CandidateType type) {
this(cid, host, jid, port, type.type_preference() + local_priority, type);
}
public Candidate.proxy(string cid, Socks5Bytestreams.Proxy proxy, int local_priority) {
this.build(cid, proxy.host, proxy.jid, proxy.port, local_priority, CandidateType.PROXY);
}
public static Candidate parse(StanzaNode candidate) throws Jingle.IqError {
string? cid = candidate.get_attribute("cid");
string? host = candidate.get_attribute("host");
string? jid_str = candidate.get_attribute("jid");
Jid? jid = jid_str != null ? Jid.parse(jid_str) : null;
int port = candidate.get_attribute("port") != null ? candidate.get_attribute_int("port") : 1080;
int priority = candidate.get_attribute_int("priority");
string? type_str = candidate.get_attribute("type");
CandidateType type = type_str != null ? CandidateType.parse(type_str) : CandidateType.DIRECT;
if (cid == null || host == null || jid == null || port <= 0 || priority <= 0) {
throw new Jingle.IqError.BAD_REQUEST("missing or invalid cid, host, jid or port");
}
return new Candidate(cid, host, jid, port, priority, type);
}
public StanzaNode to_xml() {
return new StanzaNode.build("candidate", NS_URI)
.put_attribute("cid", cid)
.put_attribute("host", host)
.put_attribute("jid", jid.to_string())
.put_attribute("port", port.to_string())
.put_attribute("priority", priority.to_string())
.put_attribute("type", type_.to_string());
}
}
bool bytes_equal(uint8[] a, uint8[] b) {
if (a.length != b.length) {
return false;
}
for (int i = 0; i < a.length; i++) {
if (a[i] != b[i]) {
return false;
}
}
return true;
}
class Parameters : Jingle.TransportParameters, Object {
public Jingle.Role role { get; private set; }
public string sid { get; private set; }
public string remote_dstaddr { get; private set; }
public string local_dstaddr { get; private set; }
public Gee.List<Candidate> local_candidates = new ArrayList<Candidate>();
public Gee.List<Candidate> remote_candidates = new ArrayList<Candidate>();
Jid local_full_jid;
Jid peer_full_jid;
bool remote_sent_selected_candidate = false;
Candidate? remote_selected_candidate = null;
bool local_determined_selected_candidate = false;
Candidate? local_selected_candidate = null;
SocketConnection? local_selected_candidate_conn = null;
weak Jingle.Session? session = null;
XmppStream? hack = null;
string? waiting_for_activation_cid = null;
SourceFunc waiting_for_activation_callback;
bool waiting_for_activation_error = false;
private static string calculate_dstaddr(string sid, Jid first_jid, Jid second_jid) {
string hashed = sid + first_jid.to_string() + second_jid.to_string();
return Checksum.compute_for_string(ChecksumType.SHA1, hashed);
}
private Parameters(Jingle.Role role, string sid, Jid local_full_jid, Jid peer_full_jid, string? remote_dstaddr) {
this.role = role;
this.sid = sid;
this.local_dstaddr = calculate_dstaddr(sid, local_full_jid, peer_full_jid);
this.remote_dstaddr = remote_dstaddr ?? calculate_dstaddr(sid, peer_full_jid, local_full_jid);
this.local_full_jid = local_full_jid;
this.peer_full_jid = peer_full_jid;
}
public Parameters.create(Jid local_full_jid, Jid peer_full_jid, string sid) {
this(Jingle.Role.INITIATOR, sid, local_full_jid, peer_full_jid, null);
}
public static Parameters parse(Jid local_full_jid, Jid peer_full_jid, StanzaNode transport) throws Jingle.IqError {
string? dstaddr = transport.get_attribute("dstaddr");
string? mode = transport.get_attribute("mode");
string? sid = transport.get_attribute("sid");
if (mode != null && mode != "tcp") {
throw new Jingle.IqError.BAD_REQUEST(@"unknown transport method $(mode)");
}
if (dstaddr != null && dstaddr.length > 255) {
throw new Jingle.IqError.BAD_REQUEST("too long dstaddr");
}
Parameters result = new Parameters(Jingle.Role.RESPONDER, sid, local_full_jid, peer_full_jid, dstaddr);
foreach (StanzaNode candidate in transport.get_subnodes("candidate", NS_URI)) {
result.remote_candidates.add(Candidate.parse(candidate));
}
return result;
}
public string transport_ns_uri() {
return NS_URI;
}
public StanzaNode to_transport_stanza_node() {
StanzaNode transport = new StanzaNode.build("transport", NS_URI)
.add_self_xmlns()
.put_attribute("dstaddr", local_dstaddr);
if (role == Jingle.Role.INITIATOR) {
// Must not be included by the responder according to XEP-0260.
transport.put_attribute("mode", "tcp");
}
transport.put_attribute("sid", sid);
foreach (Candidate candidate in local_candidates) {
transport.put_node(candidate.to_xml());
}
return transport;
}
public void on_transport_accept(StanzaNode transport) throws Jingle.IqError {
Parameters other = Parameters.parse(local_full_jid, peer_full_jid, transport);
if (other.sid != sid) {
throw new Jingle.IqError.BAD_REQUEST("invalid sid");
}
remote_candidates = other.remote_candidates;
remote_dstaddr = other.remote_dstaddr;
}
public void on_transport_info(StanzaNode transport) throws Jingle.IqError {
StanzaNode? candidate_error = transport.get_subnode("candidate-error", NS_URI);
StanzaNode? candidate_used = transport.get_subnode("candidate-used", NS_URI);
StanzaNode? activated = transport.get_subnode("activated", NS_URI);
StanzaNode? proxy_error = transport.get_subnode("proxy-error", NS_URI);
int num_children = 0;
if (candidate_error != null) { num_children += 1; }
if (candidate_used != null) { num_children += 1; }
if (activated != null) { num_children += 1; }
if (proxy_error != null) { num_children += 1; }
if (num_children == 0) {
throw new Jingle.IqError.UNSUPPORTED_INFO("unknown transport-info");
} else if (num_children > 1) {
throw new Jingle.IqError.BAD_REQUEST("transport-info with more than one child");
}
if (candidate_error != null) {
handle_remote_candidate(null);
}
if (candidate_used != null) {
string? cid = candidate_used.get_attribute("cid");
if (cid == null) {
throw new Jingle.IqError.BAD_REQUEST("missing cid");
}
handle_remote_candidate(cid);
}
if (activated != null) {
string? cid = activated.get_attribute("cid");
if (cid == null) {
throw new Jingle.IqError.BAD_REQUEST("missing cid");
}
handle_activated(cid);
}
if (proxy_error != null) {
handle_proxy_error();
}
}
private void handle_remote_candidate(string? cid) throws Jingle.IqError {
if (remote_sent_selected_candidate) {
throw new Jingle.IqError.BAD_REQUEST("remote candidate already specified");
}
Candidate? candidate = null;
if (cid != null) {
foreach (Candidate c in local_candidates) {
if (c.cid == cid) {
candidate = c;
break;
}
}
if (candidate == null) {
throw new Jingle.IqError.BAD_REQUEST("unknown cid");
}
}
remote_sent_selected_candidate = true;
remote_selected_candidate = candidate;
try_completing_negotiation();
}
private void handle_activated(string cid) throws Jingle.IqError {
if (waiting_for_activation_cid == null || cid != waiting_for_activation_cid) {
throw new Jingle.IqError.BAD_REQUEST("unexpected proxy activation message");
}
Idle.add((owned)waiting_for_activation_callback);
waiting_for_activation_cid = null;
}
private void handle_proxy_error() throws Jingle.IqError {
if (waiting_for_activation_cid == null) {
throw new Jingle.IqError.BAD_REQUEST("unexpected proxy error message");
}
Idle.add((owned)waiting_for_activation_callback);
waiting_for_activation_cid = null;
waiting_for_activation_error = true;
}
private void try_completing_negotiation() {
if (!remote_sent_selected_candidate || !local_determined_selected_candidate) {
return;
}
Candidate? remote = remote_selected_candidate;
Candidate? local = local_selected_candidate;
int num_candidates = 0;
if (remote != null) { num_candidates += 1; }
if (local != null) { num_candidates += 1; }
if (num_candidates == 0) {
// Notify Jingle of the failed transport.
session.set_transport_connection(hack, null);
return;
}
bool remote_wins;
if (num_candidates == 1) {
remote_wins = remote != null;
} else {
if (local.priority < remote.priority) {
remote_wins = true;
} else if (local.priority > remote.priority) {
remote_wins = false;
} else {
// equal priority -> XEP-0260 says that the candidate offered
// by the initiator wins, so the one that the remote chose
remote_wins = role == Jingle.Role.INITIATOR;
}
}
if (!remote_wins) {
if (local_selected_candidate.type_ != CandidateType.PROXY) {
Jingle.Session? strong = session;
if (strong == null) {
return;
}
strong.set_transport_connection(hack, local_selected_candidate_conn);
} else {
wait_for_remote_activation.begin(local_selected_candidate, local_selected_candidate_conn);
}
} else {
connect_to_local_candidate.begin(remote_selected_candidate);
}
}
public async void wait_for_remote_activation(Candidate candidate, SocketConnection conn) {
waiting_for_activation_cid = candidate.cid;
waiting_for_activation_callback = wait_for_remote_activation.callback;
yield;
Jingle.Session? strong = session;
if (strong == null) {
return;
}
if (!waiting_for_activation_error) {
strong.set_transport_connection(hack, conn);
} else {
strong.set_transport_connection(hack, null);
}
}
public async void connect_to_local_candidate(Candidate candidate) {
try {
SocketConnection conn = yield connect_to_socks5(candidate, local_dstaddr);
bool activation_error = false;
SourceFunc callback = connect_to_local_candidate.callback;
StanzaNode query = new StanzaNode.build("query", Socks5Bytestreams.NS_URI)
.add_self_xmlns()
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("activate", Socks5Bytestreams.NS_URI)
.put_node(new StanzaNode.text(peer_full_jid.to_string()))
);
Iq.Stanza iq = new Iq.Stanza.set(query) { to=candidate.jid };
hack.get_module(Iq.Module.IDENTITY).send_iq(hack, iq, (stream, iq) => {
activation_error = iq.is_error();
Idle.add((owned)callback);
});
yield;
if (activation_error) {
throw new IOError.PROXY_FAILED("activation iq error");
}
Jingle.Session? strong = session;
if (strong == null) {
return;
}
strong.send_transport_info(hack, new StanzaNode.build("transport", NS_URI)
.add_self_xmlns()
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("activated", NS_URI)
.put_attribute("cid", candidate.cid)
)
);
strong.set_transport_connection(hack, conn);
} catch (Error e) {
Jingle.Session? strong = session;
if (strong == null) {
return;
}
strong.send_transport_info(hack, new StanzaNode.build("transport", NS_URI)
.add_self_xmlns()
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("proxy-error", NS_URI))
);
strong.set_transport_connection(hack, null);
}
}
public async SocketConnection connect_to_socks5(Candidate candidate, string dstaddr) throws Error {
SocketClient socket_client = new SocketClient() { timeout=3 };
string address = @"[$(candidate.host)]:$(candidate.port)";
size_t written;
size_t read;
uint8[] read_buffer = new uint8[1024];
ByteArray write_buffer = new ByteArray();
SocketConnection conn = yield socket_client.connect_to_host_async(address, 0);
// 05 SOCKS version 5
// 01 number of authentication methods: 1
// 00 nop authentication
yield conn.output_stream.write_all_async({0x05, 0x01, 0x00}, GLib.Priority.DEFAULT, null, out written);
yield conn.input_stream.read_all_async(read_buffer[0:2], GLib.Priority.DEFAULT, null, out read);
// 05 SOCKS version 5
// 01 success
if (read_buffer[0] != 0x05 || read_buffer[1] != 0x00) {
throw new IOError.PROXY_FAILED("wanted 05 00, got %02x %02x".printf(read_buffer[0], read_buffer[1]));
}
// 05 SOCKS version 5
// 01 connect
// 00 reserved
// 03 address type: domain name
// ?? length of the domain
// .. domain
// 00 port 0 (upper half)
// 00 port 0 (lower half)
write_buffer.append({0x05, 0x01, 0x00, 0x03});
write_buffer.append({(uint8)dstaddr.length});
write_buffer.append(dstaddr.data);
write_buffer.append({0x00, 0x00});
yield conn.output_stream.write_all_async(write_buffer.data, GLib.Priority.DEFAULT, null, out written);
yield conn.input_stream.read_all_async(read_buffer[0:write_buffer.len], GLib.Priority.DEFAULT, null, out read);
// 05 SOCKS version 5
// 00 success
// 00 reserved
// 03 address type: domain name
// ?? length of the domain
// .. domain
// 00 port 0 (upper half)
// 00 port 0 (lower half)
if (read_buffer[0] != 0x05 || read_buffer[1] != 0x00 || read_buffer[3] != 0x03) {
throw new IOError.PROXY_FAILED("wanted 05 00 ?? 03, got %02x %02x %02x %02x".printf(read_buffer[0], read_buffer[1], read_buffer[2], read_buffer[3]));
}
if (read_buffer[4] != (uint8)dstaddr.length) {
throw new IOError.PROXY_FAILED("wanted %02x for length, got %02x".printf(dstaddr.length, read_buffer[4]));
}
if (!bytes_equal(read_buffer[5:5+dstaddr.length], dstaddr.data)) {
string repr = ((string)read_buffer[5:5+dstaddr.length]).make_valid().escape();
throw new IOError.PROXY_FAILED(@"wanted dstaddr $(dstaddr), got $(repr)");
}
if (read_buffer[5+dstaddr.length] != 0x00 || read_buffer[5+dstaddr.length+1] != 0x00) {
throw new IOError.PROXY_FAILED("wanted port 00 00, got %02x %02x".printf(read_buffer[5+dstaddr.length], read_buffer[5+dstaddr.length+1]));
}
conn.get_socket().set_timeout(0);
return conn;
}
public async void try_connecting_to_candidates(XmppStream stream, Jingle.Session session) throws Error {
remote_candidates.sort((c1, c2) => {
// sort from priorities from high to low
if (c1.priority < c2.priority) { return 1; }
if (c1.priority > c2.priority) { return -1; }
return 0;
});
foreach (Candidate candidate in remote_candidates) {
if (remote_selected_candidate != null && remote_selected_candidate.priority > candidate.priority) {
// Don't try candidates with lower priority than the one the
// peer already selected.
break;
}
try {
SocketConnection conn = yield connect_to_socks5(candidate, remote_dstaddr);
local_determined_selected_candidate = true;
local_selected_candidate = candidate;
local_selected_candidate_conn = conn;
session.send_transport_info(stream, new StanzaNode.build("transport", NS_URI)
.add_self_xmlns()
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("candidate-used", NS_URI)
.put_attribute("cid", candidate.cid)
)
);
try_completing_negotiation();
return;
} catch (Error e) {
// An error in the connection establishment isn't fatal, just
// try the next candidate or respond that none of the
// candidates work.
}
}
local_determined_selected_candidate = true;
local_selected_candidate = null;
session.send_transport_info(stream, new StanzaNode.build("transport", NS_URI)
.add_self_xmlns()
.put_attribute("sid", sid)
.put_node(new StanzaNode.build("candidate-error", NS_URI))
);
}
public void create_transport_connection(XmppStream stream, Jingle.Session session) {
this.session = session;
this.hack = stream;
try_connecting_to_candidates.begin(stream, session);
}
}
}

35
xmpp-vala/src/module/xep/0261_jingle_in_band_bytestreams.vala
View file

@ -30,28 +30,38 @@ public class Module : Jingle.Transport, XmppStreamModule {
public Jingle.TransportType transport_type() {
return Jingle.TransportType.STREAMING;
}
public Jingle.TransportParameters create_transport_parameters() {
return new Parameters(random_uuid(), DEFAULT_BLOCKSIZE);
public int transport_priority() {
return 0;
}
public Jingle.TransportParameters parse_transport_parameters(StanzaNode transport) throws Jingle.IqError {
return Parameters.parse(transport);
public Jingle.TransportParameters create_transport_parameters(XmppStream stream, Jid local_full_jid, Jid peer_full_jid) {
return new Parameters.create(peer_full_jid, random_uuid());
}
public Jingle.TransportParameters parse_transport_parameters(XmppStream stream, Jid local_full_jid, Jid peer_full_jid, StanzaNode transport) throws Jingle.IqError {
return Parameters.parse(peer_full_jid, transport);
}
}
class Parameters : Jingle.TransportParameters, Object {
public Jingle.Role role { get; private set; }
public Jid peer_full_jid { get; private set; }
public string sid { get; private set; }
public int block_size { get; private set; }
public Parameters(string sid, int block_size) {
private Parameters(Jingle.Role role, Jid peer_full_jid, string sid, int block_size) {
this.role = role;
this.peer_full_jid = peer_full_jid;
this.sid = sid;
this.block_size = block_size;
}
public static Parameters parse(StanzaNode transport) throws Jingle.IqError {
public Parameters.create(Jid peer_full_jid, string sid) {
this(Jingle.Role.INITIATOR, peer_full_jid, sid, DEFAULT_BLOCKSIZE);
}
public static Parameters parse(Jid peer_full_jid, StanzaNode transport) throws Jingle.IqError {
string? sid = transport.get_attribute("sid");
int block_size = transport.get_attribute_int("block-size");
if (sid == null || block_size <= 0 || block_size > MAX_BLOCKSIZE) {
throw new Jingle.IqError.BAD_REQUEST("missing or invalid sid or blocksize");
}
return new Parameters(sid, block_size);
return new Parameters(Jingle.Role.RESPONDER, peer_full_jid, sid, block_size);
}
public string transport_ns_uri() {
return NS_URI;
@ -62,15 +72,18 @@ class Parameters : Jingle.TransportParameters, Object {
.put_attribute("block-size", block_size.to_string())
.put_attribute("sid", sid);
}
public void update_transport(StanzaNode transport) throws Jingle.IqError {
Parameters other = Parameters.parse(transport);
public void on_transport_accept(StanzaNode transport) throws Jingle.IqError {
Parameters other = Parameters.parse(peer_full_jid, transport);
if (other.sid != sid || other.block_size > block_size) {
throw new Jingle.IqError.NOT_ACCEPTABLE("invalid IBB sid or block_size");
}
block_size = other.block_size;
}
public IOStream create_transport_connection(XmppStream stream, Jid peer_full_jid, Jingle.Role role) {
return InBandBytestreams.Connection.create(stream, peer_full_jid, sid, block_size, role == Jingle.Role.INITIATOR);
public void on_transport_info(StanzaNode transport) throws Jingle.IqError {
throw new Jingle.IqError.UNSUPPORTED_INFO("transport-info not supported for IBBs");
}
public void create_transport_connection(XmppStream stream, Jingle.Session session) {
session.set_transport_connection(stream, InBandBytestreams.Connection.create(stream, peer_full_jid, sid, block_size, role == Jingle.Role.INITIATOR));
}
}

4
xmpp-vala/src/module/xep/0313_message_archive_management.vala
View file

@ -39,7 +39,7 @@ public class Module : XmppStreamModule {
}
StanzaNode query_node = new StanzaNode.build("query", NS_VER(stream)).add_self_xmlns().put_node(data_form.get_submit_node());
Iq.Stanza iq = new Iq.Stanza.set(query_node);
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { page_through_results(stream, iq, on_finished); });
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { page_through_results(stream, iq, (owned)on_finished); });
}
public override void attach(XmppStream stream) {
@ -69,7 +69,7 @@ public class Module : XmppStreamModule {
)
)
);
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, paging_iq, (stream, iq) => { page_through_results(stream, iq, on_finished); });
stream.get_module(Iq.Module.IDENTITY).send_iq(stream, paging_iq, (stream, iq) => { page_through_results(stream, iq, (owned)on_finished); });
}
private void query_availability(XmppStream stream) {