From e6feefa1fc257c76d20d09295bd263a6e7f69e66 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 18 Aug 2014 17:55:26 +0100 Subject: [PATCH] initial commit --- CONTRIBUTORS | 5 + LICENSE | 191 ++++++++++++++ NOTICE | 3 + README.md | 214 ++++++++++++++++ connection.go | 568 ++++++++++++++++++++++++++++++++++++++++++ connection_maker.go | 142 +++++++++++ consts.go | 46 ++++ crypto.go | 449 +++++++++++++++++++++++++++++++++ docs/architecture.txt | 209 ++++++++++++++++ docs/testing.txt | 161 ++++++++++++ ethernet_decoder.go | 119 +++++++++ forwarder.go | 377 ++++++++++++++++++++++++++++ mac_cache.go | 108 ++++++++ peer.go | 386 ++++++++++++++++++++++++++++ peer_cache.go | 311 +++++++++++++++++++++++ peer_name_hash.go | 45 ++++ peer_name_mac.go | 57 +++++ router.go | 283 +++++++++++++++++++++ topology.go | 175 +++++++++++++ types.go | 274 ++++++++++++++++++++ utils.go | 95 +++++++ weave/Dockerfile | 6 + weave/Makefile | 22 ++ weave/docker-ns | 36 +++ weave/main.go | 146 +++++++++++ weave/wedo | 48 ++++ 26 files changed, 4476 insertions(+) create mode 100644 CONTRIBUTORS create mode 100644 LICENSE create mode 100644 NOTICE create mode 100644 README.md create mode 100644 connection.go create mode 100644 connection_maker.go create mode 100644 consts.go create mode 100644 crypto.go create mode 100644 docs/architecture.txt create mode 100644 docs/testing.txt create mode 100644 ethernet_decoder.go create mode 100644 forwarder.go create mode 100644 mac_cache.go create mode 100644 peer.go create mode 100644 peer_cache.go create mode 100644 peer_name_hash.go create mode 100644 peer_name_mac.go create mode 100644 router.go create mode 100644 topology.go create mode 100644 types.go create mode 100644 utils.go create mode 100644 weave/Dockerfile create mode 100644 weave/Makefile create mode 100755 weave/docker-ns create mode 100644 weave/main.go create mode 100755 weave/wedo diff --git a/CONTRIBUTORS b/CONTRIBUTORS new file mode 100644 index 00000000..ef85ad07 --- /dev/null +++ b/CONTRIBUTORS @@ -0,0 +1,5 @@ +This file lists all individuals having contributed content to the +repository. + +Matthew Sackman +Matthias Radestock diff --git a/LICENSE b/LICENSE new file mode 100644 index 00000000..ea98bb28 --- /dev/null +++ b/LICENSE @@ -0,0 +1,191 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + Copyright 2014 Zettio Ltd. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/NOTICE b/NOTICE new file mode 100644 index 00000000..97ce615c --- /dev/null +++ b/NOTICE @@ -0,0 +1,3 @@ +Weave +Copyright 2012-2014 Zettio Ltd. +This product includes software developed at Zettio Ltd. diff --git a/README.md b/README.md new file mode 100644 index 00000000..89edffdc --- /dev/null +++ b/README.md @@ -0,0 +1,214 @@ +# Weave - the Docker SDN + +Weave lets you connect Docker containers deployed across multiple +hosts, as if they were all plugged into the same network switch. + +## Installation + +Each hosts needs to have the following installed + +- `wedo` and `docker-ns` from the weave sub-directory. Install these + somewhere in the root user's path, e.g. `/usr/local/bin'. +- [pipework][], by Jérôme Petazzoni. This too needs to be installed in + the root user's path. +- `ethtool` and `arping` - grab them via your favourite package manager. + +[pipework]: https://raw.githubusercontent.com/jpetazzo/pipework/master/pipework + +(NB. The above assumes that the zettio/weave docker image is +available. Which it will be once the project is public. Meanwhile, +you'll have to [build the image yourself](#Building)). + +## Example + +Say you have docker running on two hosts, accessible to each other as +$HOST1 and $HOST2, and want to deploy an application consisting of +two containers, one on each host. + +On $HOST1 run (as root) + + host1# WEAVE=$(wedo launch 10.0.0.1/16) + host1# C=$(wedo run 10.0.1.1/24 -t -i debian /bin/bash) + +The first line starts the weave container. This needs to be done once +on each host. The second line starts our application container; this +could be any container - here we just take a stock debian container +and launch a shell in it. If our application consists of more than one +container on this host we simply launch them with a variation on that +2nd line. + +We tell weave that its IP address should be 10.0.0.1, and that the +weave network is on 10.0.0.0/16. Similarly, we specify the IP address +and network for our application container, with the network being a +subnet of the weave network. + +The IP addresses and netmasks can be anything you like which doesn't +conflict with any IP ranges of 'external' services your containers +need to connect to. The same IP range must be used everywhere, and the +individual IP addresses must, of course, be unique. + +We repeat similar steps on $HOST2... + + host2# WEAVE=$(wedo launch 10.0.0.2/16 $HOST1) + host2# C=$(wedo run 10.0.1.2/24 -t -i debian /bin/bash) + +The only difference, apart from the IP addresses, is that we tell our +weave that it should peer with the weave running on $HOST1. We could +instead have told the weave on $HOST1 to connect to $HOST2, or told +both about each other. Order doesn't matter here; weave automatically +(re)connects to peers when they are become available. Also, we can +tell weave to connect to multiple peers by supplying multiple +addresses. + +Now that we've got everything set up, let's see whether our containers +can talk to each other... + +On $HOST1... + + host1# docker attach $C + root@28841bd02eff:/# ping -c 1 -q 10.0.1.2 + PING 10.0.1.2 (10.0.1.2): 48 data bytes + --- 10.0.1.2 ping statistics --- + 1 packets transmitted, 1 packets received, 0% packet loss + round-trip min/avg/max/stddev = 1.048/1.048/1.048/0.000 ms + +Similarly, on $HOST2... + + host2# docker attach $C + root@f76829496120:/# ping -c 1 -q 10.0.1.1 + PING 10.0.1.1 (10.0.1.1): 48 data bytes + --- 10.0.1.1 ping statistics --- + 1 packets transmitted, 1 packets received, 0% packet loss + round-trip min/avg/max/stddev = 1.034/1.034/1.034/0.000 ms + +So there we have it, two containers on separate hosts happily talking +to each other. + +## Features + +Weave has a few more features beyond those illustrated by the basic +example above... + +### Virtual Ethernet Switch + +To application containers, the network established by weave looks +like a giant Ethernet switch to which all the containers are +connected. + +That means *any* protocol is supported. Doesn't even have to be over +TCP/IP, e.g. in the above example, + + root@28841bd02eff:/# nc -lu -p 4422 + + root@f76829496120:/# echo 'Hello, world.' | nc -u 10.0.1.1 4422 + +sends some data from the 2nd container to the first over a UDP +connection on port 4422. + +We can deploy the entire arsenal of standard network tools and +applications, developed over decades, to configure, secure, monitor, +and troubleshoot our container network. To put it another way, we can +now re-use the same tools and techniques when deploying applications +as containers as we would have done when deploying them 'on metal' in +our data centre. + +### Application isolation + +A single weave network can host multiple, isolated applications, with +each application's containers being able to communicate with each +other but not containers of other applications. + +To accomplish that, we assign each application a different subnet. So, +in the above example, if we wanted to add another application similar +to, but isolated from, our first, we'd launch the containers with... + + host1# D=$(wedo run 10.0.2.1/24 -t -i debian /bin/bash) + host2# D=$(wedo run 10.0.2.2/24 -t -i debian /bin/bash) + +A quick 'ping' test in the containers confirms that they can talk to +each other but not the containers of our first application... + + host1# docker attach $D + root@da50502598d5:/# ping -c 1 -q 10.0.2.2 + PING 10.0.2.2 (10.0.2.2): 48 data bytes + --- 10.0.2.2 ping statistics --- + 1 packets transmitted, 1 packets received, 0% packet loss + round-trip min/avg/max/stddev = 0.562/0.562/0.562/0.000 ms + root@da50502598d5:/# ping -c 1 -q 10.0.1.1 + PING 10.0.1.1 (10.0.1.1) 56(84) bytes of data. + --- 10.0.1.1 ping statistics --- + 1 packets transmitted, 0 received, 100% packet loss, time 0ms + root@da50502598d5:/# ping -c 1 -q 10.0.1.2 + PING 10.0.1.2 (10.0.1.2) 56(84) bytes of data. + --- 10.0.1.2 ping statistics --- + 1 packets transmitted, 0 received, 100% packet loss, time 0ms + +This isolation-through-subnets scheme is an example of carrying over a +well-known technique from the 'on metal' days to containers. + +### Security + +In order to connect containers across untrusted networks, weave peers +can be told to encrypt traffic by supplying a `-password` option when +launching weave, e.g. + + host1# WEAVE=$(wedo launch 10.0.0.1/16 -password wEaVe) + +The same password must be specified for all weave peers. + +### Multi-cloud networking + +Weave can network containers hosted in different cloud providers / +data centres. So, for example, one could run an application consisting +of containers on GCE, AMZN and a local data centres. + +To enable this, the network must be configured to permit TCP and UDP +connections to port 6783 of the docker hosts. + +### Multi-hop routing + +A network of containers across more than two hosts can be established +even when there is only partial connectivity between the hosts. Weave +is able to route traffic between containers as long as there is at +least one *path* of connected hosts between them. + +So, for example, if a docker host in a local data centre can connect +to hosts in GCE and AMZN, but the latter two cannot connect to each +other, containers in the latter two can still communicate; weave will +route the traffic via the local data centre. + +### Dynamic topologies + +To add a host to an existing weave network, one simply launches +weave on the host, supplying the address of at least one existing +host. Weave will automatically discover the other hosts in the other +network and establish connections to them if it can (in order to avoid +unnecessary multi-hop routing). + +### Fault tolerance + +Weave peers continually exchange topology information, and monitor +and (re)establish network connections to other peers. So if hosts or +networks fail, weave can "route around" the problem. This includes +network partitions; containers on either side of a partition can +continue to communicate, with full connectivity being restored when +the partition heals. + +The weave container is very light-weight - just over 8MB image size +and a few 10s of MBs of runtime memory - and disposable. I.e. should +weave ever run into difficulty, one can simply bounce the weave +container. Application containers do *not* have to be restarted in +that event, and indeed may not even experience a temporary +connectivity failure if the weave container is restarted quickly +enough. + +## Building + +To build weave you need `libpcap-dev` and `docker` installed. + +Then simply run + + $ make -C weave + +This will build the weave application, produce a docker image +`zettio/weave` and export that image to /tmp/wedo.tar diff --git a/connection.go b/connection.go new file mode 100644 index 00000000..f3654e4d --- /dev/null +++ b/connection.go @@ -0,0 +1,568 @@ +package weave + +import ( + "encoding/binary" + "encoding/gob" + "encoding/hex" + "fmt" + "log" + "net" + "strconv" + "time" +) + +func NewRemoteConnection(from, to *Peer, tcpAddr string) *RemoteConnection { + return &RemoteConnection{ + local: from, + remote: to, + remoteTCPAddr: tcpAddr} +} + +func (conn *RemoteConnection) Local() *Peer { + return conn.local +} + +func (conn *RemoteConnection) Remote() *Peer { + return conn.remote +} + +func (conn *RemoteConnection) RemoteTCPAddr() string { + return conn.remoteTCPAddr +} + +func (conn *RemoteConnection) Shutdown() { +} + +func (conn *RemoteConnection) Established() bool { + return true +} + +func (conn *RemoteConnection) String() string { + from := "" + if conn.local != nil { + from = conn.local.Name.String() + } + to := "" + if conn.remote != nil { + to = conn.remote.Name.String() + } + return fmt.Sprint("Connection ", from, "->", to) +} + +// Async. Does not return anything. If the connection is successful, +// it will end up in the local peer's connections map. +func NewLocalConnection(connRemote *RemoteConnection, expectedName PeerName, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) { + if connRemote.local != router.Ourself { + log.Fatal("Attempt to create local connection from a peer which is not ourself") + } + + queryChan := make(chan *ConnectionInteraction, ChannelSize) + // NB, we're taking a copy of connRemote here. + connLocal := &LocalConnection{ + RemoteConnection: *connRemote, + Router: router, + TCPConn: tcpConn, + remoteUDPAddr: udpAddr, + clientPMTU: DefaultPMTU, + queryChan: queryChan} + go connLocal.queryLoop(queryChan, expectedName) +} + +func (conn *LocalConnection) Established() bool { + conn.RLock() + defer conn.RUnlock() + return conn.established +} + +// Read by the forwarder processes when in the UDP senders +func (conn *LocalConnection) RemoteUDPAddr() *net.UDPAddr { + conn.RLock() + defer conn.RUnlock() + return conn.remoteUDPAddr +} + +// Called by the forwarder processes in a few places (including +// crypto), but the connection TCP receiver process, and by the local +// peer actor process. Do not call this from the connection's actor +// process itself. +func (conn *LocalConnection) CheckFatal(err error) error { + if err == nil { + return nil + } + conn.log("encountered fatal error", err) + conn.Shutdown() + return err +} + +// Called by forwarder processes, read in Forward (by sniffer and udp +// listener process in router). +func (conn *LocalConnection) setClientPMTU(pmtu int) { + conn.Lock() + defer conn.Unlock() + conn.clientPMTU = pmtu + conn.pmtuVerified = false + conn.log("Client PMTU set to", pmtu) +} + +func (conn *LocalConnection) clientPMTUVerification(pmtu int) { + conn.Lock() + defer conn.Unlock() + if pmtu == conn.clientPMTU && !conn.pmtuVerified { + conn.log("Client PMTU verified at", pmtu) + conn.pmtuVerified = true + } +} + +func (conn *LocalConnection) isClientPMTUVerfied() bool { + conn.RLock() + defer conn.RUnlock() + return conn.pmtuVerified +} + +// Called by the connection's actor process, by the connection's +// heartbeat process and by the connection's TCP received +// process. StackFrag is read in conn.Forward (called by router udp +// listener and sniffer processes) +func (conn *LocalConnection) setStackFrag(frag bool) { + conn.Lock() + defer conn.Unlock() + conn.stackFrag = frag +} + +func (conn *LocalConnection) log(args ...interface{}) { + // why oh why can't I just write + // log.Println("Connection to peer", conn.remote.Name, args...)? + v := append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name)) + v = append(v, args...) + log.Println(v...) +} + +// ACTOR client API + +const ( + CSendTCP = iota + CSetEstablished = iota + CSetRemoteUDPAddr = iota + CShutdown = iota +) + +// Async +func (conn *LocalConnection) Shutdown() { + conn.queryChan <- &ConnectionInteraction{Interaction: Interaction{code: CShutdown}} +} + +// Async +func (conn *LocalConnection) SetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) { + if remoteUDPAddr == nil { + return + } + conn.queryChan <- &ConnectionInteraction{ + Interaction: Interaction{code: CSetRemoteUDPAddr}, + payload: remoteUDPAddr} +} + +// Async +func (conn *LocalConnection) SetEstablished() { + conn.queryChan <- &ConnectionInteraction{Interaction: Interaction{code: CSetEstablished}} +} + +// Async +func (conn *LocalConnection) SendTCP(msg []byte) { + conn.queryChan <- &ConnectionInteraction{ + Interaction: Interaction{code: CSendTCP}, + payload: msg} +} + +// ACTOR server + +func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, expectedName PeerName) { + err := conn.handshake(expectedName) + if err != nil { + log.Printf("->[%s] encountered error during handshake: %v\n", conn.remoteTCPAddr, err) + conn.handleShutdown() + return + } + conn.local.AddConnection(conn) + if conn.remoteUDPAddr != nil { + err = conn.ensureHeartbeat(true) + } + terminate := false + for !terminate { + if err != nil { + conn.log("encountered fatal error", err) + break + } + query, ok := <-queryChan + if !ok { + break + } + switch { + case query.code == CShutdown: + terminate = true + case query.code == CSetEstablished: + err = conn.handleSetEstablished() + case query.code == CSetRemoteUDPAddr: + err = conn.handleSetRemoteUDPAddr(query.payload.(*net.UDPAddr)) + case query.code == CSendTCP: + err = conn.handleSendTCP(query.payload.([]byte)) + } + } + conn.handleShutdown() +} + +func (conn *LocalConnection) handleSetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) error { + if remoteUDPAddr == nil { + return nil + } + conn.Lock() + old := conn.remoteUDPAddr + conn.remoteUDPAddr = remoteUDPAddr + conn.Unlock() + if old == nil { + if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil { + return err + } + return conn.handleSetEstablished() + } else if old.String() != remoteUDPAddr.String() { + log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) + } + return nil +} + +func (conn *LocalConnection) handleSetEstablished() error { + conn.Lock() + old := conn.established + conn.established = true + conn.Unlock() + if !old { + conn.local.ConnectionEstablished(conn) + if err := conn.ensureHeartbeat(false); err != nil { + return err + } + // Send a large frame down the DF channel in order to prompt + // PMTU discovery to start. + conn.Forward(true, &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: PMTUDiscovery}, + nil) + conn.setStackFrag(false) + return conn.handleSendTCP(ProtocolStartFragmentationTestByte) + } + return nil +} + +func (conn *LocalConnection) handleSendTCP(msg []byte) error { + return conn.tcpSender.Send(msg) +} + +func (conn *LocalConnection) handleShutdown() { + if conn.remote != nil { + conn.log("shutting down") + } + + // Whilst some of these elements may have been written to whilst + // holding locks, they were only written to by the connection + // actor process. handleShutdown is only called by the connection + // actor process. So there is no need to take locks to read these + // (or write elements which are only read by the same + // process). Taking locks is only done for elements which are read + // by other processes. + if conn.shutdown { + return + } + conn.shutdown = true + + if conn.TCPConn != nil { + checkWarn(conn.TCPConn.Close()) + } + + if conn.remote != nil { + conn.remote.DecrementLocalRefCount() + conn.local.DeleteConnection(conn) + } + + if conn.heartbeatStop != nil { + // heartbeatStop is 0 length, so this send will synchronise + // with the receive. + conn.heartbeatStop <- nil + } + + // blank out the forwardChan so that the router processes don't + // try to send any more + conn.stopForwarders() + + if conn.remote != nil { + conn.Router.ConnectionMaker.EnsureConnection(conn.remote.Name, conn.remoteTCPAddr) + } +} + +func (conn *LocalConnection) handshake(expectedName PeerName) error { + // We do not need to worry about locking in here as at this point, + // the connection is not reachable by any go-routine other than + // ourself. Only when we add this connection to the conn.local + // peer will it be visible from multiple go-routines. + tcpConn := conn.TCPConn + tcpConn.SetKeepAlive(true) + tcpConn.SetLinger(0) + tcpConn.SetNoDelay(true) + + enc := gob.NewEncoder(tcpConn) + + localConnID := randUint64() + + versionStr := fmt.Sprint(ProtocolVersion) + handshakeSend := map[string]string{ + "Protocol": Protocol, + "ProtocolVersion": versionStr, + "PeerNameFlavour": PeerNameFlavour, + "Name": conn.local.Name.String(), + "UID": fmt.Sprint(conn.local.UID), + "ConnID": fmt.Sprint(localConnID)} + handshakeRecv := map[string]string{} + + usingPassword := conn.Router.UsingPassword() + var public, private *[32]byte + var err error + if usingPassword { + public, private, err = GenerateKeyPair() + if err != nil { + return err + } + handshakeSend["PublicKey"] = hex.EncodeToString(public[:]) + } + enc.Encode(handshakeSend) + + dec := gob.NewDecoder(tcpConn) + err = dec.Decode(&handshakeRecv) + if err != nil { + return err + } + _, err = checkHandshakeStringField("Protocol", Protocol, handshakeRecv) + if err != nil { + return err + } + _, err = checkHandshakeStringField("ProtocolVersion", versionStr, handshakeRecv) + if err != nil { + return err + } + _, err = checkHandshakeStringField("PeerNameFlavour", PeerNameFlavour, handshakeRecv) + if err != nil { + return err + } + nameStr, err := checkHandshakeStringField("Name", "", handshakeRecv) + if err != nil { + return err + } + name, err := PeerNameFromString(nameStr) + if err != nil { + return err + } + if expectedName != UnknownPeerName && expectedName != name { + return fmt.Errorf("Unexpected remote name: expected %s, received %s", expectedName, name) + } + + uidStr, err := checkHandshakeStringField("UID", "", handshakeRecv) + if err != nil { + return err + } + uid, err := strconv.ParseUint(uidStr, 10, 64) + if err != nil { + return err + } + + remoteConnIdStr, err := checkHandshakeStringField("ConnID", "", handshakeRecv) + if err != nil { + return err + } + remoteConnID, err := strconv.ParseUint(remoteConnIdStr, 10, 64) + if err != nil { + return err + } + conn.UID = localConnID ^ remoteConnID + + if usingPassword { + remotePublicStr, rpErr := checkHandshakeStringField("PublicKey", "", handshakeRecv) + if rpErr != nil { + return rpErr + } + remotePublicSlice, rpErr := hex.DecodeString(remotePublicStr) + if rpErr != nil { + return rpErr + } + remotePublic := [32]byte{} + for idx, elem := range remotePublicSlice { + remotePublic[idx] = elem + } + conn.SessionKey = FormSessionKey(&remotePublic, private, conn.Router.Password) + conn.tcpSender = NewEncryptedTCPSender(enc, conn) + conn.Decryptor = NewNaClDecryptor(conn) + } else { + if _, found := handshakeRecv["PublicKey"]; found { + return fmt.Errorf("Remote network is encrypted. Password required.") + } + conn.tcpSender = NewSimpleTCPSender(enc) + conn.Decryptor = NewNonDecryptor(conn) + } + + toPeer := NewPeer(name, uid, 0, conn.Router) + toPeer = conn.Router.Peers.FetchWithDefault(toPeer) + if toPeer == nil { + return fmt.Errorf("Connection appears to be with different version of a peer we already know of") + } else if toPeer == conn.local { + // have to do assigment here to ensure Shutdown releases ref count + conn.remote = toPeer + return fmt.Errorf("Cannot connect to ourself") + } + conn.remote = toPeer + + go conn.receiveTCP(dec, usingPassword) + return nil +} + +func checkHandshakeStringField(fieldName string, expectedValue string, handshake map[string]string) (string, error) { + val, found := handshake[fieldName] + if !found { + return "", fmt.Errorf("Field % is missing", fieldName) + } + if expectedValue != "" && val != expectedValue { + return "", fmt.Errorf("Field %s has wrong value; expected '%s', received '%s'", fieldName, expectedValue, val) + } + return val, nil +} + +func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder, usingPassword bool) { + defer conn.Decryptor.Shutdown() + var receiver TCPReceiver + if usingPassword { + receiver = NewEncryptedTCPReceiver(conn) + } else { + receiver = NewSimpleTCPReceiver() + } + var err error + for { + var msg []byte + if conn.CheckFatal(decoder.Decode(&msg)) != nil { + return + } + msg, err = receiver.Decode(msg) + if err != nil { + checkWarn(err) + // remote peer may be using wrong password. Or there could + // be some sort of injection attack going on. Just ignore + // the traffic rather than shutting down. + continue + } + if msg[0] == ProtocolConnectionEstablished { + // We initiated the connection. We sent fast heartbeats to + // the remote side, which has now received at least one of + // them and thus has informed us via TCP that it considers + // the connection is now up. We now do a fetchAll on it. + conn.SetEstablished() + conn.SendTCP(ProtocolFetchAllByte) + } else if msg[0] == ProtocolStartFragmentationTest { + conn.Forward(false, &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: FragTest}, + nil) + } else if msg[0] == ProtocolFragmentationReceived { + conn.setStackFrag(true) + } else if usingPassword && msg[0] == ProtocolNonce { + conn.Decryptor.ReceiveNonce(msg[1:]) + } else if msg[0] == ProtocolFetchAll { + // There are exactly two messages that relate to topology + // updates. + // + // 1. FetchAll. This carries no payload. The receiver + // responds with the entire topology model as the receiver + // has it. + // + // 2. Update. This carries a topology payload. The + // receiver merges it with its own topology model. If the + // payload is a subset of the receiver's topology, no + // further action is taken. Otherwise, the receiver sends + // out to all its connections an "improved" update: + // - elements which the original payload added to the + // receiver are included + // - elements which the original payload updated in the + // receiver are included + // - elements which are equal between the receiver and + // the payload are not included + // - elements where the payload was older than the + // receiver's version are updated + conn.SendTCP(Concat(ProtocolUpdateByte, conn.Router.Topology.FetchAll())) + } else if msg[0] == ProtocolUpdate { + newUpdate, err := conn.Router.Peers.ApplyUpdate(msg[1:], conn.Router) + if _, ok := err.(UnknownPeersError); err != nil && ok { + checkWarn(err) + conn.SendTCP(ProtocolFetchAllByte) + continue + } else if conn.CheckFatal(err) != nil { + return + } + if len(newUpdate) == 0 { + continue + } + conn.local.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate)) + } else if msg[0] == ProtocolPMTUVerified { + conn.clientPMTUVerification(int(binary.BigEndian.Uint16(msg[1:]))) + } else { + conn.log("received unknown msg:\n", msg) + } + } +} + +// Heartbeats + +// Heartbeating serves two purposes: a) keeping NAT paths alive, and +// b) updating a remote peer's knowledge of our address, in the event +// it changes (e.g. because NAT paths expired). +// Called only by connection actor process. +func (conn *LocalConnection) ensureHeartbeat(fast bool) error { + if err := conn.ensureForwarders(); err != nil { + return err + } + var heartbeat, fetchAll, fragTest <-chan time.Time + // explicitly 0 length chan - make send block until receive occurs + stop := make(chan interface{}, 0) + if fast { + // fast, nofetchall, no fragtest + // Lang Spec: "A nil channel is never ready for communication." + heartbeat = time.Tick(FastHeartbeat) + } else { + heartbeat = time.Tick(SlowHeartbeat) + fetchAll = time.Tick(FetchAllInterval) + fragTest = time.Tick(FragTestInterval) + } + // Don't need locks here as this is only read here and in + // handleShutdown, both of which are called by the connection + // actor process only. + if conn.heartbeatStop != nil { + conn.heartbeatStop <- nil + } + conn.heartbeatStop = stop + go conn.forwardHeartbeats(heartbeat, fetchAll, fragTest, stop) + return nil +} + +func (conn *LocalConnection) forwardHeartbeats(heartbeat, fetchAll, fragTest <-chan time.Time, stop <-chan interface{}) { + heartbeatFrame := &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: []byte{}} + conn.Forward(true, heartbeatFrame, nil) // avoid initial wait + for { + select { + case <-stop: + return + case <-heartbeat: + conn.Forward(true, heartbeatFrame, nil) + case <-fetchAll: + conn.SendTCP(ProtocolFetchAllByte) + case <-fragTest: + conn.setStackFrag(false) + conn.SendTCP(ProtocolStartFragmentationTestByte) + } + } +} diff --git a/connection_maker.go b/connection_maker.go new file mode 100644 index 00000000..94d1cf3c --- /dev/null +++ b/connection_maker.go @@ -0,0 +1,142 @@ +package weave + +import ( + "bytes" + "fmt" + "log" + "math/rand" + "net" + "strconv" + "time" +) + +const ( + InitialInterval = 5 * time.Second + MaxInterval = 10 * time.Minute + MaxAttemptCount = 100 + CMEnsure = iota + CMStatus = iota +) + +func StartConnectionMaker(router *Router) *ConnectionMaker { + queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) + state := &ConnectionMaker{ + router: router, + queryChan: queryChan, + failedConnections: make(map[PeerName]*FailedConnection)} + go state.queryLoop(queryChan) + return state +} + +func (cm *ConnectionMaker) EnsureConnection(name PeerName, foundAt string) { + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMEnsure}, + name: name, + foundAt: foundAt} +} + +func (cm *ConnectionMaker) String() string { + resultChan := make(chan interface{}, 0) + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMStatus, resultChan: resultChan}} + result := <-resultChan + return result.(string) +} + +func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteraction) { + var tick <-chan time.Time + maybeTick := func() { + if tick == nil && len(cm.failedConnections) > 0 { + tick = time.After(5 * time.Second) + } + } + for { + select { + case query, ok := <-queryChan: + if !ok { + return + } + switch { + case query.code == CMEnsure: + if query.name != cm.router.Ourself.Name { + cm.addToFailedConnection(query.name, query.foundAt) + maybeTick() + } + case query.code == CMStatus: + query.resultChan <- cm.status() + default: + log.Fatal("Unexpected connection maker query:", query) + } + case now := <-tick: + for name, failedConn := range cm.failedConnections { + if now.After(failedConn.tryAfter) { + if _, found := cm.router.Ourself.ConnectionTo(name); found { + delete(cm.failedConnections, name) + continue + } else if failedConn.attemptCount == MaxAttemptCount { + delete(cm.failedConnections, name) + continue + } + after, interval := tryAfter(failedConn.tryInterval) + failedConn.tryInterval = interval + failedConn.tryAfter = after + failedConn.attemptCount += 1 + for target := range failedConn.foundAt { + go cm.attemptConnection(target, name) + } + } + } + tick = nil + maybeTick() + } + } +} + +func (cm *ConnectionMaker) addToFailedConnection(name PeerName, foundAt string) { + failed := cm.failedConnections[name] + if failed == nil { + after, interval := tryAfter(InitialInterval) + failed = &FailedConnection{ + foundAt: make(map[string]bool), + tryInterval: interval, + tryAfter: after} + } + foundAtHost, foundAtPortStr, err := net.SplitHostPort(foundAt) + if err == nil { + // ensure port-less version is there + failed.foundAt[foundAtHost] = true + if foundAtPort, err := strconv.Atoi(foundAtPortStr); err == nil && foundAtPort != Port { + failed.foundAt[foundAt] = true + } + } else { + // can't split it, assume it must not have port on it + failed.foundAt[foundAt] = true + } + cm.failedConnections[name] = failed +} + +func (cm *ConnectionMaker) status() string { + var buf bytes.Buffer + for name, failedConn := range cm.failedConnections { + foundAt := make([]string, 0, len(failedConn.foundAt)) + for target := range failedConn.foundAt { + foundAt = append(foundAt, target) + } + buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt)) + } + return buf.String() +} + +func (cm *ConnectionMaker) attemptConnection(foundAt string, targetName PeerName) { + if err := cm.router.Ourself.CreateConnection(foundAt, targetName); err != nil { + log.Println(err) + } +} + +func tryAfter(interval time.Duration) (time.Time, time.Duration) { + interval += time.Duration(rand.Int63n(int64(interval))) + if interval > MaxInterval { + interval = MaxInterval + } + return time.Now().Add(interval), interval +} diff --git a/consts.go b/consts.go new file mode 100644 index 00000000..e5cf40b5 --- /dev/null +++ b/consts.go @@ -0,0 +1,46 @@ +package weave + +import ( + "time" +) + +const ( + Protocol = "weave" + ProtocolVersion = 8 + UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP + EthernetOverhead = 14 + Port = 6783 + DefaultPMTU = 65535 + MaxUDPPacketSize = 65536 + ChannelSize = 16 + UDPNonceSendAt = 8192 + FragTestSize = 60001 + PMTUDiscoverySize = 60000 + FastHeartbeat = 500 * time.Millisecond + SlowHeartbeat = 10 * time.Second + FetchAllInterval = 30 * time.Second + FragTestInterval = 5 * time.Minute + PMTUVerifyTimeout = 5 * time.Second +) + +const ( + ProtocolConnectionEstablished = iota + ProtocolFragmentationReceived = iota + ProtocolStartFragmentationTest = iota + ProtocolNonce = iota + ProtocolFetchAll = iota + ProtocolUpdate = iota + ProtocolPMTUVerified = iota +) + +var ( + FragTest = make([]byte, FragTestSize) + PMTUDiscovery = make([]byte, PMTUDiscoverySize) + ProtocolConnectionEstablishedByte = []byte{ProtocolConnectionEstablished} + ProtocolFragmentationReceivedByte = []byte{ProtocolFragmentationReceived} + ProtocolStartFragmentationTestByte = []byte{ProtocolStartFragmentationTest} + ProtocolNonceByte = []byte{ProtocolNonce} + ProtocolFetchAllByte = []byte{ProtocolFetchAll} + ProtocolUpdateByte = []byte{ProtocolUpdate} + ProtocolPMTUVerifiedByte = []byte{ProtocolPMTUVerified} +) diff --git a/crypto.go b/crypto.go new file mode 100644 index 00000000..14c868ae --- /dev/null +++ b/crypto.go @@ -0,0 +1,449 @@ +package weave + +import ( + "bytes" + "code.google.com/p/go-bit/bit" + "code.google.com/p/go.crypto/nacl/box" + "code.google.com/p/go.crypto/nacl/secretbox" + "crypto/rand" + "crypto/sha256" + "encoding/binary" + "encoding/gob" + "fmt" + "log" +) + +func GenerateKeyPair() (publicKey, privateKey *[32]byte, err error) { + return box.GenerateKey(rand.Reader) +} + +func FormSessionKey(remotePublicKey, localPrivateKey *[32]byte, secretKey *[]byte) *[32]byte { + var sharedKey [32]byte + box.Precompute(&sharedKey, remotePublicKey, localPrivateKey) + sharedKeySlice := sharedKey[:] + sharedKeySlice = append(sharedKeySlice, *secretKey...) + sessionKey := sha256.Sum256(sharedKeySlice) + return &sessionKey +} + +func GenerateRandomNonce() ([24]byte, error) { + var nonce [24]byte + n, err := rand.Read(nonce[:]) + if err != nil { + return nonce, err + } + if n != 24 { + return nonce, fmt.Errorf("Did not read enough - wanted 24, got %v", n) + } + return nonce, nil +} + +func EncryptPrefixNonce(plaintxt []byte, nonce *[24]byte, secret *[32]byte) []byte { + buf := make([]byte, 24, 24+len(plaintxt)+secretbox.Overhead) + copy(buf, nonce[:]) + // Seal *appends* to buf + return secretbox.Seal(buf, plaintxt, nonce, secret) +} + +func DecryptPrefixNonce(ciphertxt []byte, secret *[32]byte) ([]byte, bool) { + if len(ciphertxt) < secretbox.Overhead+24 { + return nil, false + } + // There is no way to nicely convert from a slice to an + // array. So have to used the following loop. + var nonce [24]byte + for idx, e := range ciphertxt[0:24] { + nonce[idx] = e + } + ciphertxt = ciphertxt[24:] + return secretbox.Open(nil, ciphertxt, &nonce, secret) +} + +func SetNonceLow15Bits(nonce *[24]byte, offset uint16) { + // ensure top bit of offset is 0 + offset = offset & ((1 << 15) - 1) + // grab top bit of nonce[22:24] (and clear out lower bits) + nonceBits := binary.BigEndian.Uint16(nonce[22:24]) & (1 << 15) + // Big endian => the MSB is stored at the *lowest* address. So + // that top bit in nonce[22] should stay as the top bit in + // nonce[22] + binary.BigEndian.PutUint16(nonce[22:24], nonceBits+offset) +} + +// Nonce encoding/decoding + +func EncodeNonce(df bool) (*[24]byte, []byte, error) { + nonce, err := GenerateRandomNonce() + if err != nil { + return nil, []byte{}, err + } + // wipe out lowest 15 bits, but encode the df right at the bottom + flags := uint16(0) + if df { + flags = flags | 1 + } + SetNonceLow15Bits(&nonce, flags) + return &nonce, Concat(ProtocolNonceByte, nonce[:]), nil +} + +func DecodeNonce(msg []byte) (bool, *[24]byte) { + flags := uint16(msg[23]) + df := 0 != (flags & 1) + nonce := [24]byte{} + // upper bound is exclusive so this avoids copying the flags + for idx, elem := range msg[:23] { + nonce[idx] = elem + } + return df, &nonce +} + +// Frame Encryptors + +func NewNonEncryptor(prefix []byte) *NonEncryptor { + buf := make([]byte, MaxUDPPacketSize) + prefixLen := copy(buf, prefix) + return &NonEncryptor{ + buf: buf, + bufTail: buf[prefixLen:], + buffered: prefixLen, + prefixLen: prefixLen} +} + +func (ne *NonEncryptor) PacketOverhead() int { + return ne.prefixLen +} + +func (ne *NonEncryptor) FrameOverhead() int { + return NameSize + NameSize + 2 +} + +func (ne *NonEncryptor) IsEmpty() bool { + return ne.buffered == ne.prefixLen +} + +func (ne *NonEncryptor) Bytes() []byte { + buf := ne.buf[:ne.buffered] + ne.buffered = ne.prefixLen + ne.bufTail = ne.buf[ne.prefixLen:] + return buf +} + +func (ne *NonEncryptor) AppendFrame(frame *ForwardedFrame) { + bufTail := ne.bufTail + srcLen := copy(bufTail, frame.srcPeer.NameByte) + bufTail = bufTail[srcLen:] + dstLen := copy(bufTail, frame.dstPeer.NameByte) + bufTail = bufTail[dstLen:] + frameLen := len(frame.frame) + binary.BigEndian.PutUint16(bufTail, uint16(frameLen)) + bufTail = bufTail[2:] + copy(bufTail, frame.frame) + ne.bufTail = bufTail[frameLen:] + ne.buffered = ne.buffered + srcLen + dstLen + 2 + frameLen +} + +func (ne *NonEncryptor) TotalLen() int { + return ne.buffered +} + +func NewNaClEncryptor(prefix []byte, conn *LocalConnection, df bool) *NaClEncryptor { + buf := make([]byte, MaxUDPPacketSize) + prefixLen := copy(buf, prefix) + flags := uint16(0) + if df { + flags = flags | (1 << 15) + } + return &NaClEncryptor{ + NonEncryptor: *NewNonEncryptor([]byte{}), + buf: buf, + offset: 0, + nonce: nil, + nonceChan: make(chan *[24]byte, ChannelSize), + flags: flags, + prefixLen: prefixLen, + conn: conn, + df: df} +} + +func (ne *NaClEncryptor) Bytes() []byte { + plaintext := ne.NonEncryptor.Bytes() + offsetFlags := ne.offset | ne.flags + ciphertext := ne.buf + binary.BigEndian.PutUint16(ciphertext[ne.prefixLen:], offsetFlags) + nonce := ne.nonce + if nonce == nil { + freshNonce, encodedNonce, err := EncodeNonce(ne.df) + if err = ne.conn.CheckFatal(err); err != nil { + return []byte{} + } + ne.conn.SendTCP(encodedNonce) + ne.nonce = freshNonce + nonce = freshNonce + } + offset := ne.offset + SetNonceLow15Bits(nonce, offset) + // Seal *appends* to ciphertext + ciphertext = secretbox.Seal(ciphertext[:ne.prefixLen+2], plaintext, nonce, ne.conn.SessionKey) + + offset = (offset + 1) & ((1 << 15) - 1) + if offset == 0 { + // need a new nonce please + ne.nonce = <-ne.nonceChan + } else if offset == 1<<14 { // half way through range, send new nonce + nonce, encodedNonce, err := EncodeNonce(ne.df) + if err = ne.conn.CheckFatal(err); err != nil { + return []byte{} + } + ne.nonceChan <- nonce + ne.conn.SendTCP(encodedNonce) + } + ne.offset = offset + + return ciphertext +} + +func (ne *NaClEncryptor) PacketOverhead() int { + return ne.prefixLen + 2 + secretbox.Overhead + ne.NonEncryptor.PacketOverhead() +} + +func (ne *NaClEncryptor) TotalLen() int { + return ne.PacketOverhead() + ne.NonEncryptor.TotalLen() +} + +// Frame Decryptors + +func NewNonDecryptor(conn *LocalConnection) *NonDecryptor { + return &NonDecryptor{conn: conn} +} + +func (nd *NonDecryptor) IterateFrames(fun FrameConsumer, packet *UDPPacket) error { + buf := packet.Packet + for len(buf) >= (2 + NameSize + NameSize) { + srcNameByte := buf[:NameSize] + buf = buf[NameSize:] + dstNameByte := buf[:NameSize] + buf = buf[NameSize:] + length := binary.BigEndian.Uint16(buf[:2]) + frame := buf[2 : 2+length] + buf = buf[2+length:] + err := fun(nd.conn, packet.Sender, srcNameByte, dstNameByte, length, frame) + if err != nil { + return err + } + } + return nil +} + +func (nd *NonDecryptor) Shutdown() { +} + +func (nd *NonDecryptor) ReceiveNonce(msg []byte) { + log.Println("Received Nonce on non-encrypted channel. Ignoring.") +} + +func NewNaClDecryptor(conn *LocalConnection) *NaClDecryptor { + inst := NaClDecryptorInstance{ + nonce: nil, + previousNonce: nil, + usedOffsets: bit.New(), + previousUsedOffsets: nil, + highestOffsetSeen: 0, + nonceChan: make(chan *[24]byte, ChannelSize)} + instDF := NaClDecryptorInstance{ + nonce: nil, + previousNonce: nil, + usedOffsets: bit.New(), + previousUsedOffsets: nil, + highestOffsetSeen: 0, + nonceChan: make(chan *[24]byte, ChannelSize)} + return &NaClDecryptor{ + NonDecryptor: *NewNonDecryptor(conn), + instance: &inst, + instanceDF: &instDF} +} + +func (nd *NaClDecryptor) Shutdown() { + close(nd.instance.nonceChan) + close(nd.instanceDF.nonceChan) +} + +func (nd *NaClDecryptor) ReceiveNonce(msg []byte) { + df, nonce := DecodeNonce(msg) + if df { + nd.instanceDF.nonceChan <- nonce + } else { + nd.instance.nonceChan <- nonce + } +} + +func (nd *NaClDecryptor) IterateFrames(fun FrameConsumer, packet *UDPPacket) error { + buf, err := nd.decrypt(packet.Packet) + if err != nil { + return err + } + packet.Packet = buf + return nd.NonDecryptor.IterateFrames(fun, packet) +} + +func (nd *NaClDecryptor) decrypt(buf []byte) ([]byte, error) { + offset := binary.BigEndian.Uint16(buf[:2]) + df := (offset & (1 << 15)) != 0 + offsetNoFlags := offset & ((1 << 15) - 1) + var decState *NaClDecryptorInstance + if df { + decState = nd.instanceDF + } else { + decState = nd.instance + } + var nonce *[24]byte + var usedOffsets *bit.Set + var ok bool + if decState.nonce == nil { + if offsetNoFlags > (1 << 13) { + // offset is already beyond the first quarter and it's the + // first thing we've seen?! I don't think so. + return nil, fmt.Errorf("Unexpected offset when decrypting UDP packet") + } + decState.nonce, ok = <-decState.nonceChan + if !ok { + return nil, fmt.Errorf("Nonce chan closed") + } + nonce = decState.nonce + usedOffsets = decState.usedOffsets + decState.highestOffsetSeen = offsetNoFlags + } else { + highestOffsetSeen := decState.highestOffsetSeen + if offsetNoFlags < (1<<13) && highestOffsetSeen > ((1<<14)+(1<<13)) && + (highestOffsetSeen-offsetNoFlags) > ((1<<14)+(1<<13)) { + // offset is in the first quarter, highestOffsetSeen is in + // the top quarter and under a quarter behind us. We + // interpret this as we need to move to the next nonce + decState.previousUsedOffsets = decState.usedOffsets + decState.usedOffsets = bit.New() + decState.previousNonce = decState.nonce + decState.nonce, ok = <-decState.nonceChan + if !ok { + return nil, fmt.Errorf("Nonce chan closed") + } + decState.highestOffsetSeen = offsetNoFlags + nonce = decState.nonce + usedOffsets = decState.usedOffsets + } else if offsetNoFlags > highestOffsetSeen && + (offsetNoFlags-highestOffsetSeen) < (1<<13) { + // offset is under a quarter above highestOffsetSeen. This + // is ok - maybe some packet loss + decState.highestOffsetSeen = offsetNoFlags + nonce = decState.nonce + usedOffsets = decState.usedOffsets + } else if offsetNoFlags <= highestOffsetSeen && + (highestOffsetSeen-offsetNoFlags) < (1<<13) { + // offset is within a quarter of the highest we've + // seen. This is ok - just assuming some out-of-order + // delivery. + nonce = decState.nonce + usedOffsets = decState.usedOffsets + } else if highestOffsetSeen < (1<<13) && offsetNoFlags > ((1<<14)+(1<<13)) && + (offsetNoFlags-highestOffsetSeen) > ((1<<14)+(1<<13)) { + // offset is in the last quarter, highestOffsetSeen is in + // the first quarter, and offset is under a quarter behind + // us. This is ok - as above, just some out of order. But + // here it means we're dealing with the previous nonce + nonce = decState.previousNonce + usedOffsets = decState.previousUsedOffsets + } else { + return nil, fmt.Errorf("Unexpected offset when decrypting UDP packet") + } + } + offsetNoFlagsInt := int(offsetNoFlags) + if usedOffsets.Contains(offsetNoFlagsInt) { + return nil, fmt.Errorf("Suspected replay attack detected when decrypting UDP packet") + } + SetNonceLow15Bits(nonce, offsetNoFlags) + result, success := secretbox.Open(nil, buf[2:], nonce, nd.conn.SessionKey) + if success { + usedOffsets.Add(offsetNoFlagsInt) + return result, nil + } else { + return nil, fmt.Errorf("Unable to decrypt msg via UDP: %v", buf) + } +} + +// TCP Senders + +func NewSimpleTCPSender(encoder *gob.Encoder) *SimpleTCPSender { + return &SimpleTCPSender{encoder: encoder} +} + +func (sender *SimpleTCPSender) Send(msg []byte) error { + return sender.encoder.Encode(msg) +} + +func NewEncryptedTCPSender(encoder *gob.Encoder, conn *LocalConnection) *EncryptedTCPSender { + buffer := new(bytes.Buffer) + return &EncryptedTCPSender{ + outerEncoder: encoder, + innerEncoder: gob.NewEncoder(buffer), + buffer: buffer, + conn: conn, + msgCount: 0} +} + +func (sender *EncryptedTCPSender) Send(msg []byte) error { + nonce, err := GenerateRandomNonce() + if err != nil { + return err + } + sender.Lock() + defer sender.Unlock() + wrappedMsg := EncryptedTCPMessage{Number: sender.msgCount, Body: msg} + buffer := sender.buffer + buffer.Reset() + err = sender.innerEncoder.Encode(wrappedMsg) + if err != nil { + return err + } + sender.msgCount = sender.msgCount + 1 + return sender.outerEncoder.Encode( + EncryptPrefixNonce(buffer.Bytes(), &nonce, sender.conn.SessionKey)) +} + +// TCP Receivers + +func NewSimpleTCPReceiver() *SimpleTCPReceiver { + return &SimpleTCPReceiver{} +} + +func (receiver *SimpleTCPReceiver) Decode(msg []byte) ([]byte, error) { + return msg, nil +} + +func NewEncryptedTCPReceiver(conn *LocalConnection) *EncryptedTCPReceiver { + buffer := new(bytes.Buffer) + return &EncryptedTCPReceiver{ + conn: conn, + decoder: gob.NewDecoder(buffer), + buffer: buffer, + msgCount: 0} +} + +func (receiver *EncryptedTCPReceiver) Decode(msg []byte) ([]byte, error) { + plaintext, success := DecryptPrefixNonce(msg, receiver.conn.SessionKey) + if !success { + return msg, fmt.Errorf("Unable to decrypt msg via TCP:\n %X", msg) + } + receiver.buffer.Reset() + _, err := receiver.buffer.Write(plaintext) + if err != nil { + return msg, err + } + wrappedMsg := new(EncryptedTCPMessage) + err = receiver.decoder.Decode(wrappedMsg) + if err != nil { + return msg, err + } + if wrappedMsg.Number != receiver.msgCount { + return msg, fmt.Errorf("Received TCP message with wrong sequence number; possible replay attack") + } + receiver.msgCount = receiver.msgCount + 1 + return wrappedMsg.Body, nil +} diff --git a/docs/architecture.txt b/docs/architecture.txt new file mode 100644 index 00000000..3f6480dd --- /dev/null +++ b/docs/architecture.txt @@ -0,0 +1,209 @@ +Router: + - Starts Topology + - Starts Connection Maker + - Starts TCP Listener + - Starts UDP Listener + - Sniffs traffic + - Starts the local peer + +- UDP Listener + This is a process that reads off the UDP socket, decodes the frames + and does either relaying to a connected peer (or peers), or + injection to the local interface (or both) (router.listenUDP / + router.handleUDPPacketFunc) + +- Traffic Sniffer + This is a process that captures packets with pcap, decodes the + frames and forwards them to a local peer (or peers) (router.Sniff / + router.handleCapturedPacket) + +- Topology + Maintains unicast and broadcast routing tables which can be read + directly from any thread by just obtaining a readlock on topology. + - Unicast answers the question "If I want to send a packet to peer + X, which of the peers I'm connected to should I send the packet + to?" + - Broadcast answers the question "If peer X broadcasts a packet, by + the time it reaches me, which of my connected peers is X expecting + me to pass the packet to?" + Also spawns a thread which runs an actor loop, responding to + requests to encode all peers, or rebuild the routing tables. + +- Connection Maker + Spawns a thread which runs an actor loop. This actor loop is passed + the known locations of peers as they're discovered and is informed + when connections we have made die. It proactively tries to create + connections to remote peers we're not connected to using a random + binary exponential backoff to regulate the period between connection + attempts. + +- Local Peer + This is the representation of the local peer. The local peer is only + peer that is "active" in any running weave. Inactive peers simply + get given some state such their name, version, UID and + connections. These are set when a network update is received from a + remote peer. This state can be read directly whilst holding a read + lock regardless of whether the peer is local or not. + + The local active peer spawns an actor thread which is mainly used by + the local connections. The actor thread manages state changes + relating to the local connections, identifying duplicate + connections, managing broadcasting network updates when its set of + connections (and hence version) changes. + + The local peer is directly called by the router traffic sniffer and + router udp listener processes to send traffic to neighbouring + peers. This is done directly with read locks. In these methods, we + inspect the unicast and broadcast routing tables as generated by + Topology. + +- Connections + The connection life cycle is as follows: + 1. Either a TCP connection is created to, or received from, the remote peer + 2. We spawn a thread which will eventually become the connection actor loop + 3. This thread starts by doing the TCP handshake directly + 4. Assuming this succeeds, a TCP receiver thread is spawned. This + simply responds to traffic received from the remote peer via TCP + 5. We register this connection with the local peer + 6a. If we initiated this connection then we now start sending fast + heartbeats (a new thread is spawned to do the heartbeating) to the + remote peer so that the remote peer can determine what + address/port it should use to send UDP back to us. To do this, we + also spawn off two further threads which are the forwarder + loops. These receive frames which are to be sent to the remote + peer. One of them sends frames without the DF flag set. To do this + it just sends the packets out of the UDP Listener socket. The + other needs to send frames with the DF flag set and needs its own + socket so that it can do PMTU discovery easily. To do this, it + uses a Raw IP socket (IP has no ports, so there's no collision + issue with the UDP Listener socket) and so it must add UDP headers + itself. + 6b. If we did not initiate this connection then the UDP Listener + should start receiving fast heartbeats from the remote peer. From + those it should be able to identify the local connection via the + local peer. It will tell the local connection (communicating to + the actor thread) about the UDP address of the remote peer. The + local connection will then start its forwarder threads as + described in 6a, and start sending slow heartbeats (heartbeating + is done by another process). The local connection is marked as + established which means it is included in network updates + broadcast to our peers. We also send to the remote peer via TCP a + ConnectionEstablished message. The remote peer receives this (on + the TCP receiver process), tells the connection actor process, + which then replaces the fast heartbeater with a slow heartbeater + and similarly marks the connection as established. + 7. Whenever a connection is established or terminated, the local + peer's version is incremented. Whenever this happens, the peer + generates a network update message which is broadcast to its + directly connected neighbours via TCP. This network update message + contains the relevant changes to the network topology due to the + connection change. When such a message is received by a TCP + receiver threads, they invoke the peers cache to apply the update + to the local model of the network topology. This may fail for a + number of reasons (for example the update may contain reference to + peers that we have no prior knowledge of. In this case, we ignore + the update and send back to the peer from whom we received the + update a request for the complete network topology), or it may + apply and elicit some changes to our model. If it does elicit some + changes then we send out an updated update message to all our + peers. In this way the changes are passed quickly from peer to + peer based on the established connections and stop being sent once + a received update causes no changes to the peer's topology + model. Changes are additive and care is taken to ensure that no + loops can occur. + + +Routing + +In general, we build up knowledge by looking only at layer 2 +(ethernet) MAC addresses. Packets that we sniff from the local +interface must have MAC address of local interfaces thus we associate +such MAC addresses with ourself. Packets that we receive via UDP from +other peers have the association embedded in the UDP traffic so that +we can add the association of the src MAC with the originating +peer. Even when a packet is relayed by an intermediate peer, we +maintain the information as to who originally sniffed the packet so +that all peers can build up the same set of associations. + +If we were implementing a hub, it would be legal to just broadcast all +packets to everyone. That would work, but it would be wasteful. As +we're more implementing a switch, if we know the destination of a +packet then we will form a packet which includes the frame sniffed, +our own identity as the original sniffer of the packet and the +destination peer identity (so that we don't rely on intermediate peers +having the same knowledge as us as to which MACs are where). We then +consult the routing tables to determine which of our connections we +should use in order to try and get the packet to its ultimate +destination peer. The packet we form does not include any information +as to the route we expect the packet to take - we merely determine the +next hop and entrust that to do the onwards routing. Any intermediate +peer will receive the packet, should be able to identify the +destination peer and then similarly consult its own routing tables to +determine the next hop. The intermediate peer does not need to know in +advance the association between the destination MAC and the +destination peer. Intermediate peers do however decode the frame +sufficiently to add the association of the source peer and the source +MAC. + + +PMTU + +PMTU is the lowest MTU of hop in the path between two different +nodes. In general, it is a benefit to know what the PMTU is so that +you can perform any necessary fragmentation of packets at the +endpoints of the path to avoid any refragmentation needing to +occur. If you allow refrafmentation to occur then you can end up with +many small packets and network performance can suffer. To discover +PMTU, there is an IP flag called "Don't Fragment" (DF). With this set, +if a node receives a packet that is bigger than the next MTU, it is +required to drop the packet and send back an ICMP 3,4 packet which +informs the sending node of the next hop MTU (RFC 1191). In theory, +these ICMP packets should have all the reverse NAT and so forth +applied, to make it all the way back to the sender. In practise this +does often work, but in some cases all ICMP are sometimes blocked, +often by idiots who don't understand that ICMP is really the error +channel of all network traffic. + +Because of the encapsulation overheads, it is important that weave +respects PMTU. If it sniffs a packet of X bytes, the weave-to-weave +traffic will be some N bytes bigger than X. If the packet happens to +be an IP packet and it happens to have DF set then we should set DF on +the larger packet we send between weaves. If X+N > the PMTU between +weave peers then our send will error. We will then hopefully be able +to query what the actual PMTU is, and then can subtract N from this +and send an resultant ICMP3,4 packet back to the original sender. + +It is frequently the case that large UDP packets that don't have DF +set (either the sending side chose not to set DF or the packet is +greater than the PMTU so it can't have DF set) get dropped. Ideally, +large packets without DF set should just get transparently fragmented +and reassembled and there should be no packet loss but in reality this +often doesn't occur. Therefore weave tests to see whether or not +fragmentation of large UDP packets between peers is reliable or not, +and retests from time to time. + +If weave determines that fragmentation is reliable then when weave +sniffs large packets (i.e. packets larger than the MTU and thus don't +have DF set) then weave will encapsulate these packets as necessary +and similarly send them out as-is, without DF set, trusting to the +network to do all necessary fragementation and reassembly. + +However, if weave determines that fragementation is not reliable +between any two peers then it will manually fragment larger packets +correctly according to the IP spec, and will then send them between +weave peers with the DF flag set (i.e. the fragmentation will ensure +that the encapsulatated traffic will not be greater than the PMTU +between weave peers). Because the fragmentation is done according to +the IP spec, we don't need to do reassembly ourselves - on the +receiving weave, we jus inject all the fragments and rely on the +stack to do reassembly. + +Sometimes PMTU discovery doesn't work as ICMP packets sometimes get +dropped by firewalls. Whenever weave sends between peers with DF set +and gets an error informing it of a new PMTU, it will attempt to +verify that PMTU by sending packets of exactly that size to the remote +peer. If it gets no indication from the remote weave that these +packets have been received within a timeout period, it will +artificially drop the PMTU by 10 and try again. Thus this will slowly +reduce the PMTU until a working PMTU is found. This is slow, but it +does work and works around ICMP packets being dropped. diff --git a/docs/testing.txt b/docs/testing.txt new file mode 100644 index 00000000..8bea09c2 --- /dev/null +++ b/docs/testing.txt @@ -0,0 +1,161 @@ +http://baturin.org/docs/iproute2/ is a very useful cheat-sheet for the +"ip" command generally (especially the bit on namespaces). + +The following assumes you have two VMs set up on your computer and +they are both on the same bridge. + +From the host, the bridge (virbr0) has IP 192.168.122.1 +VM1 has eth0 at 192.168.122.2 +VM2 has eth0 at 192.168.122.3 + +All 3 can ping each other and generally contact each other. You do not +need to worry about forwarding on the host as all traffic will stay on +this bridge and not leave it. + +The following work for me, but there seem to be a variety of bugs and +weirdnesses in the networking stack so the following is a bit odd... + +Setting up VM1: + +VM1> brctl addbr br0 # create bridge +VM1> ip netns add ns1 # create ns1 +VM1> ip netns add ns2 # create ns2 +VM1> ip link add name veth1 type veth peer name veth2 # create veth pair veth1/veth2 +VM1> ip link add name veth3 type veth peer name veth4 # create veth pair veth3/veth4 +VM1> ip link add name veth5 type veth peer name veth6 # create veth pair veth5/veth6 +VM1> ip link set dev veth2 netns ns1 # move veth2 into ns1 +VM1> ip link set dev veth4 netns ns2 # move veth4 into ns2 +VM1> ip netns exec ns1 ip link set dev veth2 up # switch on veth2 +VM1> ip netns exec ns2 ip link set dev veth4 up # switch on veth4 +VM1> ip link set dev veth6 up # switch on veth6 +VM1> ifconfig br0 10.0.0.1/24 up # configure and enable br0 +VM1> ip netns exec ns1 ip address add 10.0.0.2/24 dev veth2 # configure veth2 +VM1> ip netns exec ns2 ip address add 10.0.0.3/24 dev veth4 # configure veth4 +VM1> ip address add 10.0.0.4/24 dev veth6 +VM1> ip link set dev veth1 up # switch on veth1 +VM1> ip link set dev veth3 up # switch on veth3 +VM1> ip link set dev veth5 up # switch on veth5 +VM1> brctl addif br0 veth1 # add veth1 to br0 +VM1> brctl addif br0 veth3 # add veth3 to br0 +VM1> brctl addif br0 veth5 # add veth5 to br0 + +Now we need to turn off checksum offloading on sending interfaces: + +VM1> ip netns exec ns1 ethtool -K veth2 tx off +VM1> ip netns exec ns2 ethtool -K veth4 tx off + +We've created 3 veth pairs, but only put an end of two of them inside +name spaces. We're not going to run weave inside a namespace. We're +instead going to run it on veth6. To avoid confusing the kernel +routing tables, we need to remove an entry it will have added: + +VM1> route del -net 10.0.0.0/24 dev veth6 + +The kernel should (and will) use br0 as the entry point to our +10.0.0.0/24 network, not veth6. + +At this point, you should be able to successfully ping 10.0.0.1, +10.0.0.2, 10.0.0.3 and 10.0.0.4 from VM1. But you can also try pinging +from inside the two namespaces (ns1 and ns2). + +VM1> ping -c1 10.0.0.1 +VM1> ping -c1 10.0.0.2 +VM1> ping -c1 10.0.0.3 +VM1> ping -c1 10.0.0.4 +VM1> ip netns exec ns1 ping -c1 10.0.0.1 +VM1> ip netns exec ns1 ping -c1 10.0.0.2 +VM1> ip netns exec ns1 ping -c1 10.0.0.3 +VM1> ip netns exec ns1 ping -c1 10.0.0.4 +etc + +In these VMs, we need the kernel to forward traffic between interfaces +(outside of bridges) so we need forwarding on: + +VM1> echo 1 > /proc/sys/net/ipv4/ip_forward + +Setting up VM2: + +Pretty much the same as VM1, but we use 10.0.0.11, 10.0.0.12, +10.0.0.13 and 10.0.0.14 instead: + +VM2> brctl addbr br0 +VM2> ip netns add ns1 +VM2> ip netns add ns2 +VM2> ip link add name veth1 type veth peer name veth2 +VM2> ip link add name veth3 type veth peer name veth4 +VM2> ip link add name veth5 type veth peer name veth6 +VM2> ip link set dev veth2 netns ns1 +VM2> ip link set dev veth4 netns ns2 +VM2> ip netns exec ns1 ip link set dev veth2 up +VM2> ip netns exec ns2 ip link set dev veth4 up +VM2> ip link set dev veth6 up +VM2> ifconfig br0 10.0.0.11/24 up +VM2> ip netns exec ns1 ip address add 10.0.0.12/24 dev veth2 +VM2> ip netns exec ns2 ip address add 10.0.0.13/24 dev veth4 +VM2> ip address add 10.0.0.14/24 dev veth6 +VM2> ip link set dev veth1 up +VM2> ip link set dev veth3 up +VM2> ip link set dev veth5 up +VM2> brctl addif br0 veth1 +VM2> brctl addif br0 veth3 +VM2> brctl addif br0 veth5 +VM2> echo 1 > /proc/sys/net/ipv4/ip_forward +VM2> ip netns exec ns1 ethtool --offload veth2 tx off +VM2> ip netns exec ns2 ethtool --offload veth4 tx off + +Some final checks: + +- check VM2 can't ping anything on VM1 +VM2> ping 10.0.0.1 # should fail +VM2> ip netns exec ns1 ping 10.0.0.1 # should fail +...etc... +- and vice versa + +We're going to have weave on VM2 connect to weave on VM1. Because +weave is not in any namespace, we don't need any iptables rules. + +You should now find you can initiate a connection from VM2 to VM1 on +port 6783: + +VM1> nc -l -p 6783 +VM2> nc 192.168.122.2 6783 +...type... +...type... + +If all is well, you should now be able to launch weave on VM1 +then on VM2: + +VM1> weave -iface veth6 -name vm1 +VM2> weave -iface veth6 -name vm2 192.168.122.1 + +Once the connection is up and running, you should be able to connect +from VM1/ns2 to VM2/ns2 and vice versa: + +VM1> ip netns exec ns2 nc -l -p 10000 +VM2> ip netns exec ns2 nc 10.0.0.3 10000 +...type more... +...type more... + +VM2> ip netns exec ns2 nc -l -p 10000 +VM1> ip netns exec ns2 nc 10.0.0.6 10000 +...type even more... +...type even more... + + +MTU considerations: + +MTU is the payload size of an ethernet frame. The ethernet headers are +not included in this and add a further 14 bytes (an ethernet header is +really at least 22 bytes. However only 14 bytes actually carry data - +the two mac fields and a type/length field. The other parts of the +header are there for signalling at the electrical layer and thus don't +get captured by pcap, nor need forwarding). We are then passing this +over UDP which adds a further 8 bytes, and over IP, which adds at +least another 20 bytes of headers (IP can have variable header +length. Yay). We also include the namehash as a 32 byte prefix. + +Therefore, we have 20(ip) + 8(udp) + 32(namehash) + 2(length prefix) + +14(ethernet) = 76 bytes of overhead. Thus we should make sure the +internal MTU is 76 bytes lower than the external MTU. So on a normal +1500 byte ethernet MTU, we should ensure the other containers use +nothing greater than 1424. diff --git a/ethernet_decoder.go b/ethernet_decoder.go new file mode 100644 index 00000000..870dc1bd --- /dev/null +++ b/ethernet_decoder.go @@ -0,0 +1,119 @@ +package weave + +import ( + "bytes" + "code.google.com/p/gopacket" + "code.google.com/p/gopacket/layers" + "code.google.com/p/gopacket/pcap" + "log" + "net" +) + +func NewEthernetDecoder() *EthernetDecoder { + dec := &EthernetDecoder{} + dec.parser = gopacket.NewDecodingLayerParser(layers.LayerTypeEthernet, &dec.eth, &dec.ip) + return dec +} + +func (dec *EthernetDecoder) DecodeLayers(data []byte) error { + return dec.parser.DecodeLayers(data, &dec.decoded) +} + +func (dec *EthernetDecoder) CheckFrameTooBigFunc(srcMac net.HardwareAddr, localAddrs []net.Addr, handle *pcap.Handle) func(error) error { + networks := make([]LocalAddress, len(localAddrs)) + idx := 0 + for _, addr := range localAddrs { + ip, ipnet, err := net.ParseCIDR(addr.String()) + if err == nil && ip.To4() != nil { + networks[idx] = LocalAddress{ip: ip.To4(), network: ipnet} + idx++ + } + } + networks = networks[:idx] + if len(networks) == 0 { + log.Fatal("Interface has no addresses assigned. Cannot continue") + } + findIP := func(ip net.IP) net.IP { + for _, la := range networks { + if la.network.Contains(ip) { + return la.ip + } + } + // can't find anything, we'll have to rely on default gateway. + return networks[0].ip + } + + return func(err error) error { + if ftbe, ok := err.(FrameTooBigError); ok { + // we know: 1. ip is valid, 2. it was ip and DF was set + srcIP := findIP(dec.ip.SrcIP) + icmpFrame, err := dec.formICMPMTUPacket(srcMac, srcIP, ftbe.PMTU) + if err != nil { + return err + } + log.Println("Injecting ICMP 3,4. PMTU:", ftbe.PMTU) + return handle.WritePacketData(icmpFrame) + } else { + return err + } + } +} + +func (dec *EthernetDecoder) formICMPMTUPacket(srcMac net.HardwareAddr, srcIP net.IP, mtu int) ([]byte, error) { + buf := gopacket.NewSerializeBuffer() + opts := gopacket.SerializeOptions{ + FixLengths: true, + ComputeChecksums: true} + ipHeaderSize := int(dec.ip.IHL) * 4 // IHL is the number of 32-byte words in the header + payload := gopacket.Payload(dec.ip.BaseLayer.Contents[:ipHeaderSize+8]) + err := gopacket.SerializeLayers(buf, opts, + &layers.Ethernet{ + SrcMAC: srcMac, + DstMAC: dec.eth.SrcMAC, + EthernetType: dec.eth.EthernetType}, + &layers.IPv4{ + Version: 4, + TOS: dec.ip.TOS, + Id: 0, + Flags: 0, + FragOffset: 0, + TTL: 64, + Protocol: layers.IPProtocolICMPv4, + DstIP: dec.ip.SrcIP, + SrcIP: srcIP}, + &layers.ICMPv4{ + TypeCode: 0x304, + Id: 0, + Seq: uint16(mtu)}, + &payload) + if err == nil { + return buf.Bytes(), nil + } else { + return []byte{}, err + } +} + +var ( + // see http://en.wikipedia.org/wiki/Multicast_address#Ethernet + broadcastMAC, _ = net.ParseMAC("ff:ff:ff:ff:ff:ff") + zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00") + stpMACPrefix = []byte{0x01, 0x80, 0xC2, 0x00, 0x00} + ipv4MulticastMACPrefix = []byte{0x01, 0x00, 0x5E} + ipv6MulticastMACPrefix = []byte{0x33, 0x33} +) + +func (dec *EthernetDecoder) DropFrame() bool { + return bytes.Equal(stpMACPrefix, dec.eth.DstMAC[:len(stpMACPrefix)]) +} + +func (dec *EthernetDecoder) BroadcastFrame() bool { + return bytes.Equal(broadcastMAC, dec.eth.DstMAC) || + // treat multicast frames as broadcast + bytes.Equal(ipv4MulticastMACPrefix, dec.eth.DstMAC[:len(ipv4MulticastMACPrefix)]) || + bytes.Equal(ipv6MulticastMACPrefix, dec.eth.DstMAC[:len(ipv6MulticastMACPrefix)]) +} + +func (dec *EthernetDecoder) IsPMTUVerify() bool { + return bytes.Equal(zeroMAC, dec.eth.SrcMAC) && + bytes.Equal(zeroMAC, dec.eth.DstMAC) +} diff --git a/forwarder.go b/forwarder.go new file mode 100644 index 00000000..82bbb8c2 --- /dev/null +++ b/forwarder.go @@ -0,0 +1,377 @@ +package weave + +import ( + "code.google.com/p/gopacket" + "code.google.com/p/gopacket/layers" + "log" + "net" + "syscall" + "time" +) + +func (conn *LocalConnection) ensureForwarders() error { + if conn.forwardChan != nil || conn.forwardChanDF != nil { + return nil + } + // only thing that can error, so do it earlier + rawUDPSender, err := NewRawUDPSender(conn) + if err != nil { + return err + } + + usingPassword := conn.SessionKey != nil + var encryptor, encryptorDF Encryptor + if usingPassword { + encryptor = NewNaClEncryptor(conn.local.NameByte, conn, false) + encryptorDF = NewNaClEncryptor(conn.local.NameByte, conn, true) + } else { + encryptor = NewNonEncryptor(conn.local.NameByte) + encryptorDF = NewNonEncryptor(conn.local.NameByte) + } + + // The forward chans in the conn struct are read by other + // processes, so we have to use locks. + var ( + forwardChan = make(chan *ForwardedFrame, ChannelSize) + forwardChanDF = make(chan *ForwardedFrame, ChannelSize) + stopForward = make(chan interface{}, 0) + stopForwardDF = make(chan interface{}, 0) + ) + conn.Lock() + conn.forwardChan = forwardChan + conn.forwardChanDF = forwardChanDF + conn.stopForward = stopForward + conn.stopForwardDF = stopForwardDF + conn.Unlock() + go conn.forwarderLoop(forwardChan, stopForward, encryptor, NewSimpleUDPSender(conn), DefaultPMTU) + go conn.forwarderLoop(forwardChanDF, stopForwardDF, encryptorDF, rawUDPSender, conn.clientPMTU) + + return nil +} + +func (conn *LocalConnection) stopForwarders() { + conn.Lock() + conn.forwardChan = nil + conn.forwardChanDF = nil + conn.Unlock() + // Now signal the forwarder loops to exit. They will drain the + // forwarder chans in order to unblock any router processes + // blocked on sending. + if conn.stopForward != nil { + conn.stopForward <- nil + conn.stopForwardDF <- nil + } +} + +func (conn *LocalConnection) forwarderLoop(forwardCh <-chan *ForwardedFrame, stop <-chan interface{}, enc Encryptor, udpSender UDPSender, pmtu int) { + defer udpSender.Shutdown() + // The pmtu governs the max size of msg we can pass to the + // udpSender. udpSender will add in the UDP/IP headers + var pmtuVerifyTick <-chan time.Time + var clientPMTU int + maxPayload := pmtu - UDPOverhead + updateClientPMTU := func() { + conn.setClientPMTU(clientPMTU) + pmtuVerifyFrame := &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: make([]byte, clientPMTU+EthernetOverhead)} + for cnt := 0; cnt < 10; cnt += 1 { + enc.AppendFrame(pmtuVerifyFrame) + udpSender.Send(enc.Bytes()) + } + pmtuVerifyTick = time.After(PMTUVerifyTimeout) + } + appendFrame := func(frame *ForwardedFrame) bool { + frameLen := len(frame.frame) + if enc.TotalLen()+enc.FrameOverhead()+frameLen > maxPayload { + return false + } + enc.AppendFrame(frame) + return true + } + flush := func() { + err := udpSender.Send(enc.Bytes()) + if err != nil { + if ftbe, ok := err.(FrameTooBigError); ok { + maxPayload = ftbe.PMTU - UDPOverhead + clientPMTU = maxPayload - enc.PacketOverhead() - enc.FrameOverhead() - EthernetOverhead + updateClientPMTU() + } else if PosixError(err) == syscall.ENOBUFS { + // TODO handle this better + } else { + conn.CheckFatal(err) + } + } + } + logDrop := func(frame *ForwardedFrame) { + conn.log("dropping too big frame during forwarding: frame len:", len(frame.frame), "; PMTU:", pmtu) + } + // We want to drain before exiting otherwise we could get the + // packet sniffer or udp listener blocked on sending to a full + // chan + drain := func() { + for { + select { + case <-forwardCh: + default: + return + } + } + } + var flushed, ok bool + var frame *ForwardedFrame + for { + flushed = false + select { + case <-stop: + drain() + return + case <-pmtuVerifyTick: + // We only do this case here when we know the buffers are + // all empty so that we don't risk appending verify-frames + // to other data. + pmtuVerifyTick = nil + if !conn.isClientPMTUVerfied() { + clientPMTU -= 10 + updateClientPMTU() + } + case frame = <-forwardCh: + if !appendFrame(frame) { + logDrop(frame) + continue + } + for !flushed { + select { + case frame, ok = <-forwardCh: + if !ok { + return + } + if !appendFrame(frame) { + flush() + if !appendFrame(frame) { + logDrop(frame) + flushed = true + } + } + default: + flush() + flushed = true + } + } + } + } +} + +// Called from peer.Relay[Broadcast] which is itself invoked from +// router (both UDP listener process and sniffer process). Also called +// from connection's heartbeat process, and from the connection's TCP +// receiver process. +func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *EthernetDecoder) error { + conn.RLock() + var ( + forwardChan = conn.forwardChan + forwardChanDF = conn.forwardChanDF + clientPMTU = conn.clientPMTU + stackFrag = conn.stackFrag + ) + conn.RUnlock() + + if forwardChan == nil || forwardChanDF == nil { + conn.log("Cannot forward frame yet - awaiting contact") + return nil + } + if df { + if len(frame.frame)-EthernetOverhead <= clientPMTU { + forwardChanDF <- frame + return nil + } else { + return FrameTooBigError{PMTU: clientPMTU, frame: frame} + } + } else { + if stackFrag || dec == nil || len(dec.decoded) < 2 { + forwardChan <- frame + return nil + } + // Don't have trustworthy stack, so we're going to have to + // send it DF in any case. + if len(frame.frame)-EthernetOverhead <= clientPMTU { + forwardChanDF <- frame + return nil + } + conn.Router.LogFrame("Fragmenting", frame.frame, &dec.eth) + // We can't trust the stack to fragment, we have IP, and we + // have a frame that's too big for the MTU, so we have to + // fragment it ourself. + return fragment(dec.eth, dec.ip, clientPMTU, frame, func(segFrame *ForwardedFrame) { + forwardChanDF <- segFrame + }) + } +} + +func fragment(eth layers.Ethernet, ip layers.IPv4, pmtu int, frame *ForwardedFrame, forward func(*ForwardedFrame)) error { + // We are not doing any sort of NAT, so we don't need to worry + // about checksums of IP payload (eg UDP checksum). + headerSize := int(ip.IHL) * 4 + // &^ is bit clear (AND NOT). So here we're clearing the lowest 3 + // bits. + maxSegmentSize := (pmtu - EthernetOverhead - headerSize) &^ 7 + opts := gopacket.SerializeOptions{ + FixLengths: false, + ComputeChecksums: true} + payloadSize := int(ip.Length) - headerSize + payload := ip.BaseLayer.Payload[:payloadSize] + offsetBase := int(ip.FragOffset) << 3 + origFlags := ip.Flags + ip.Flags = ip.Flags | layers.IPv4MoreFragments + ip.Length = uint16(headerSize + maxSegmentSize) + if eth.EthernetType == layers.EthernetTypeLLC { + // using LLC, so must set eth length correctly. eth length + // is just the length of the payload + eth.Length = ip.Length + } else { + eth.Length = 0 + } + for offset := 0; offset < payloadSize; offset += maxSegmentSize { + var segmentPayload []byte + if len(payload) <= maxSegmentSize { + // last one + segmentPayload = payload + ip.Length = uint16(len(payload) + headerSize) + ip.Flags = origFlags + if eth.EthernetType == layers.EthernetTypeLLC { + eth.Length = ip.Length + } else { + eth.Length = 0 + } + } else { + segmentPayload = payload[:maxSegmentSize] + payload = payload[maxSegmentSize:] + } + ip.FragOffset = uint16((offset + offsetBase) >> 3) + buf := gopacket.NewSerializeBuffer() + segPayload := gopacket.Payload(segmentPayload) + err := gopacket.SerializeLayers(buf, opts, ð, &ip, &segPayload) + if err != nil { + return err + } + // make copies of the frame we received + var segFrame ForwardedFrame = *frame + segFrame.frame = buf.Bytes() + forward(&segFrame) + } + return nil +} + +// UDP Senders + +func NewSimpleUDPSender(conn *LocalConnection) *SimpleUDPSender { + return &SimpleUDPSender{udpConn: conn.Router.UDPListener, conn: conn} +} + +func (sender *SimpleUDPSender) Send(msg []byte) error { + _, err := sender.udpConn.WriteToUDP(msg, sender.conn.RemoteUDPAddr()) + return err +} + +func (sender *SimpleUDPSender) Shutdown() error { + return nil +} + +func NewRawUDPSender(conn *LocalConnection) (*RawUDPSender, error) { + ipSocket, err := dialIP(conn) + if err != nil { + return nil, err + } + udpHeader := &layers.UDP{SrcPort: layers.UDPPort(Port)} + ipBuf := gopacket.NewSerializeBuffer() + opts := gopacket.SerializeOptions{ + FixLengths: true, + // UDP header is calculated with a phantom IP + // header. Yes, it's totally nuts. Thankfully, for UDP + // over IPv4, the checksum is optional. It's not + // optional for IPv6, but we'll ignore that for + // now. TODO + ComputeChecksums: false} + + return &RawUDPSender{ + ipBuf: ipBuf, + opts: opts, + udpHeader: udpHeader, + socket: ipSocket, + conn: conn}, nil +} + +func (sender *RawUDPSender) Send(msg []byte) error { + payload := gopacket.Payload(msg) + sender.udpHeader.DstPort = layers.UDPPort(sender.conn.RemoteUDPAddr().Port) + + err := gopacket.SerializeLayers(sender.ipBuf, sender.opts, sender.udpHeader, &payload) + if err != nil { + return err + } + packet := sender.ipBuf.Bytes() + _, err = sender.socket.Write(packet) + if err != nil { + if PosixError(err) == syscall.EMSGSIZE { + f, err := sender.socket.File() + if err != nil { + return err + } + defer f.Close() + fd := int(f.Fd()) + log.Println("EMSGSIZE on send, expecting PMTU update (IP packet was", + len(packet), "bytes, payload was", len(msg), "bytes)") + pmtu, err := syscall.GetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU) + if err != nil { + return err + } + return FrameTooBigError{PMTU: pmtu} + } else { + return err + } + } + return nil +} + +func (sender *RawUDPSender) Shutdown() error { + defer func() { sender.socket = nil }() + return sender.socket.Close() +} + +func dialIP(conn *LocalConnection) (*net.IPConn, error) { + ipLocalAddr, err := ipAddr(conn.TCPConn.LocalAddr()) + if err != nil { + return nil, err + } + ipRemoteAddr, err := ipAddr(conn.TCPConn.RemoteAddr()) + if err != nil { + return nil, err + } + ipSocket, err := net.DialIP("ip4:UDP", ipLocalAddr, ipRemoteAddr) + if err != nil { + return nil, err + } + f, err := ipSocket.File() + if err != nil { + return nil, err + } + defer f.Close() + fd := int(f.Fd()) + // This Makes sure all packets we send out have DF set on them. + err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DO) + if err != nil { + return nil, err + } + return ipSocket, nil +} + +func ipAddr(addr net.Addr) (*net.IPAddr, error) { + host, _, err := net.SplitHostPort(addr.String()) + if err != nil { + return nil, err + } + return &net.IPAddr{ + IP: net.ParseIP(host), + Zone: ""}, nil +} diff --git a/mac_cache.go b/mac_cache.go new file mode 100644 index 00000000..a387c136 --- /dev/null +++ b/mac_cache.go @@ -0,0 +1,108 @@ +package weave + +import ( + "bytes" + "fmt" + "net" + "sync" + "time" +) + +type MacCacheEntry struct { + lastSeen time.Time + peer *Peer +} + +type MacCache struct { + sync.RWMutex + table map[uint64]*MacCacheEntry + maxAge time.Duration + expiryTimer *time.Timer +} + +func NewMacCache(maxAge time.Duration) *MacCache { + res := &MacCache{ + table: make(map[uint64]*MacCacheEntry), + maxAge: maxAge} + res.setExpiryTimer() + return res +} + +func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool { + key := macint(mac) + now := time.Now() + cache.RLock() + entry, found := cache.table[key] + if found && entry.peer == peer && now.Before(entry.lastSeen.Add(cache.maxAge/10)) { + cache.RUnlock() + return false + } else { + cache.RUnlock() + } + cache.Lock() + defer cache.Unlock() + entry, found = cache.table[key] + if !found { + cache.table[key] = &MacCacheEntry{lastSeen: now, peer: peer} + return true + } + if entry.peer != peer { + entry.lastSeen = now + entry.peer = peer + return true + } + if now.After(entry.lastSeen.Add(cache.maxAge / 10)) { + entry.lastSeen = now + } + return false +} + +func (cache *MacCache) Lookup(mac net.HardwareAddr) (*Peer, bool) { + key := macint(mac) + cache.RLock() + defer cache.RUnlock() + entry, found := cache.table[key] + if !found { + return nil, false + } + return entry.peer, true +} + +func (cache *MacCache) Delete(peer *Peer) bool { + found := false + cache.Lock() + defer cache.Unlock() + for key, entry := range cache.table { + if entry.peer == peer { + delete(cache.table, key) + found = true + } + } + return found +} + +func (cache *MacCache) String() string { + var buf bytes.Buffer + cache.RLock() + defer cache.RUnlock() + for key, entry := range cache.table { + buf.WriteString(fmt.Sprintf("%v -> %s (%v)\n", intmac(key), entry.peer.Name, entry.lastSeen)) + } + return buf.String() +} + +func (cache *MacCache) setExpiryTimer() { + cache.expiryTimer = time.AfterFunc(cache.maxAge/10, func() { cache.expire() }) +} + +func (cache *MacCache) expire() { + now := time.Now() + cache.Lock() + defer cache.Unlock() + for key, entry := range cache.table { + if now.After(entry.lastSeen.Add(cache.maxAge)) { + delete(cache.table, key) + } + } + cache.setExpiryTimer() +} diff --git a/peer.go b/peer.go new file mode 100644 index 00000000..b01a3053 --- /dev/null +++ b/peer.go @@ -0,0 +1,386 @@ +package weave + +import ( + "fmt" + "log" + "net" + "sort" +) + +func NewPeer(name PeerName, uid uint64, version uint64, router *Router) *Peer { + if uid == 0 { + uid = randUint64() + } + return &Peer{ + Name: name, + NameByte: name.Bin(), + connections: make(map[PeerName]Connection), + version: version, + UID: uid, + Router: router} +} + +func (peer *Peer) Version() uint64 { + peer.RLock() + defer peer.RUnlock() + return peer.version +} + +func (peer *Peer) IncrementLocalRefCount() { + peer.Lock() + defer peer.Unlock() + peer.localRefCount += 1 +} + +func (peer *Peer) DecrementLocalRefCount() { + peer.Lock() + defer peer.Unlock() + peer.localRefCount -= 1 +} + +func (peer *Peer) IsLocallyReferenced() bool { + peer.RLock() + defer peer.RUnlock() + return peer.localRefCount != 0 +} + +func (peer *Peer) ConnectionCount() int { + peer.RLock() + defer peer.RUnlock() + return len(peer.connections) +} + +func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { + peer.RLock() + defer peer.RUnlock() + conn, found := peer.connections[name] + return conn, found // yes, you really can't inline that. FFS. +} + +func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) { + peer.RLock() + defer peer.RUnlock() + for name, conn := range peer.connections { + fun(name, conn) + } +} + +func (peer *Peer) HasPathTo(target *Peer, symmetric bool) (bool, map[PeerName]bool) { + peer.RLock() + defer peer.RUnlock() + reached := make(map[PeerName]bool) + nextWorklist := []*Peer{peer} + reached[peer.Name] = true + for len(nextWorklist) > 0 { + worklist := nextWorklist + sort.Sort(ListOfPeers(worklist)) + nextWorklist = []*Peer{} + for _, curPeer := range worklist { + if curPeer == target { + return true, reached + } + for remoteName, conn := range curPeer.connections { + if _, found := reached[remoteName]; found { + continue + } + // By taking this lock, we ensure no changes can occur + // to any peer we have reached. As a result, if/when + // we reach the target, we know the entire path from + // peer to target is valid - no changes can have been + // made to that path during our calculation. + remote := conn.Remote() + remote.RLock() + defer remote.RUnlock() + if _, found := remote.connections[curPeer.Name]; !symmetric || found { + nextWorklist = append(nextWorklist, remote) + reached[remoteName] = true + } + } + } + } + return false, reached +} + +func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { + if peer == peer.Router.Ourself { + log.Fatal("Attempt to set version and connections on the local peer", peer.Name) + } + peer.Lock() + defer peer.Unlock() + peer.version = version + peer.connections = connections + for remoteName, conn := range connections { + peer.Router.ConnectionMaker.EnsureConnection(remoteName, conn.RemoteTCPAddr()) + } +} + +func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { + return peer.Relay(peer, dstPeer, df, frame, dec) +} + +func (peer *Peer) Broadcast(df bool, frame []byte, dec *EthernetDecoder) error { + return peer.RelayBroadcast(peer, df, frame, dec) +} + +func (peer *Peer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { + relayPeerName, found := peer.Router.Topology.Unicast(dstPeer.Name) + if !found { + // Not necessarily an error as there could be a race with the + // dst disappearing whilst the frame is in flight + log.Println("Received packet for unknown destination:", dstPeer.Name) + return nil + } + conn, found := peer.ConnectionTo(relayPeerName) + if !found { + // Again, could just be a race, not necessarily an error + log.Println("Unable to find connection to relay peer", relayPeerName) + return nil + } + return conn.(*LocalConnection).Forward(df, &ForwardedFrame{ + srcPeer: srcPeer, + dstPeer: dstPeer, + frame: frame}, + dec) +} + +func (peer *Peer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { + nextHops := peer.Router.Topology.Broadcast(srcPeer.Name) + if len(nextHops) == 0 { + return nil + } + // We must not hold a read lock on peer during the conn.Forward + // below, since that is a potentially blocking operation (e.g. if + // the channel is full). + nextConns := make([]*LocalConnection, 0, len(nextHops)) + peer.RLock() + for _, hopName := range nextHops { + conn, found := peer.connections[hopName] + // Again, !found could just be due to a race. + if found { + nextConns = append(nextConns, conn.(*LocalConnection)) + } + } + peer.RUnlock() + var err error + for _, conn := range nextConns { + err = conn.Forward(df, &ForwardedFrame{ + srcPeer: srcPeer, + dstPeer: conn.Remote(), + frame: frame}, + dec) + if err != nil { + return err + } + } + return nil +} + +func (peer *Peer) String() string { + peer.RLock() + defer peer.RUnlock() + return fmt.Sprint("Peer ", peer.Name, " (v", peer.version, ") (UID ", peer.UID, ")") +} + +func (peer *Peer) StartLocalPeer() { + if peer.Router.Ourself != peer { + log.Fatal("Attempt to start peer which is not the local peer") + } + queryChan := make(chan *PeerInteraction, ChannelSize) + peer.queryChan = queryChan + go peer.queryLoop(queryChan) +} + +func (peer *Peer) CreateConnection(peerAddr string, expectedName PeerName) error { + if err := peer.checkConnectionLimit(); err != nil { + return err + } + // We're dialing the remote so that means connections will come from random ports + _, _, err := net.SplitHostPort(peerAddr) + var addrStr string + if err == nil { + addrStr = peerAddr + } else { + addrStr = fmt.Sprintf("%s:%d", peerAddr, Port) + } + tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr) + udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr) + if tcpErr != nil || udpErr != nil { + // they really should have the same value, but just in case... + if tcpErr == nil { + return udpErr + } + return tcpErr + } + tcpConn, err := net.DialTCP("tcp4", nil, tcpAddr) + if err != nil { + return err + } + connRemote := NewRemoteConnection(peer, nil, tcpConn.RemoteAddr().String()) + NewLocalConnection(connRemote, expectedName, tcpConn, udpAddr, peer.Router) + return nil +} + +// ACTOR client API + +const ( + PAddConnection = iota + PBroadcastTCP = iota + PDeleteConnection = iota + PConnectionEstablished = iota +) + +// Async: rely on the peer to shut us down if we shouldn't be adding +// ourselves, so therefore this can be async +func (peer *Peer) AddConnection(conn *LocalConnection) { + peer.queryChan <- &PeerInteraction{ + Interaction: Interaction{code: PAddConnection}, + payload: conn} +} + +// Sync. +func (peer *Peer) DeleteConnection(conn *LocalConnection) { + resultChan := make(chan interface{}) + peer.queryChan <- &PeerInteraction{ + Interaction: Interaction{code: PDeleteConnection, resultChan: resultChan}, + payload: conn} + <-resultChan +} + +// Async. +func (peer *Peer) ConnectionEstablished(conn *LocalConnection) { + peer.queryChan <- &PeerInteraction{ + Interaction: Interaction{code: PConnectionEstablished}, + payload: conn} +} + +// Async. +func (peer *Peer) BroadcastTCP(msg []byte) { + peer.queryChan <- &PeerInteraction{ + Interaction: Interaction{code: PBroadcastTCP}, + payload: msg} +} + +// ACTOR server + +func (peer *Peer) queryLoop(queryChan <-chan *PeerInteraction) { + for { + query, ok := <-queryChan + if !ok { + return + } + switch { + case query.code == PAddConnection: + peer.handleAddConnection(query.payload.(*LocalConnection)) + case query.code == PDeleteConnection: + peer.handleDeleteConnection(query.payload.(*LocalConnection)) + query.resultChan <- nil + case query.code == PConnectionEstablished: + peer.handleConnectionEstablished(query.payload.(*LocalConnection)) + case query.code == PBroadcastTCP: + peer.handleBroadcastTCP(query.payload.([]byte)) + } + } +} + +func (peer *Peer) handleAddConnection(conn *LocalConnection) { + if peer != conn.Local() { + log.Fatal("Attempt made to add connection to peer where peer is not the source of connection") + } + if conn.Remote() == nil { + log.Fatal("Attempt made to add connection to peer with unknown remote peer") + } + toName := conn.Remote().Name + dupErr := fmt.Errorf("Multiple connections to %s added to %s", toName, peer.Name) + // deliberately non symmetrical + if dupConn, found := peer.connections[toName]; found { + if dupConn == conn { + return + } else { + // conn.UID is used as the tie breaker here, in the + // knowledge that both sides will make the same decision. + dupConnLocal := dupConn.(*LocalConnection) + if conn.UID == dupConnLocal.UID { + // oh good grief. Sod it, just kill both of them. + conn.CheckFatal(dupErr) + dupConnLocal.CheckFatal(dupErr) + peer.handleDeleteConnection(dupConnLocal) + return + } else if conn.UID < dupConnLocal.UID { + dupConnLocal.CheckFatal(dupErr) + peer.handleDeleteConnection(dupConnLocal) + } else { + conn.CheckFatal(dupErr) + return + } + } + } + if err := peer.checkConnectionLimit(); err != nil { + conn.CheckFatal(err) + return + } + peer.Lock() + peer.connections[toName] = conn + peer.Unlock() +} + +func (peer *Peer) handleDeleteConnection(conn *LocalConnection) { + if peer != conn.Local() { + log.Fatal("Attempt made to delete connection from peer where peer is not the source of connection") + } + if conn.Remote() == nil { + log.Fatal("Attempt made to delete connection to peer with unknown remote peer") + } + toName := conn.Remote().Name + if connFound, found := peer.connections[toName]; !found || connFound != conn { + return + } + peer.Lock() + delete(peer.connections, toName) + peer.Unlock() + broadcast := false + if conn.Established() { + peer.Lock() + peer.version += 1 + peer.Unlock() + broadcast = true + } + // Must do garbage collection first to ensure we don't send out an + // update with unreachable peers (can cause looping) + peer.Router.Peers.GarbageCollect(peer.Router) + if broadcast { + peer.broadcastPeerUpdate() + } +} + +func (peer *Peer) handleConnectionEstablished(conn *LocalConnection) { + if peer != conn.Local() { + log.Fatal("Peer informed of active connection where peer is not the source of connection") + } + if dupConn, found := peer.connections[conn.Remote().Name]; !found || conn != dupConn { + conn.CheckFatal(fmt.Errorf("Cannot set unknown connection active")) + return + } + peer.Lock() + peer.version += 1 + peer.Unlock() + log.Println("Peer", peer.Name, "established active connection to remote peer", conn.Remote().Name, "at", conn.RemoteTCPAddr()) + peer.broadcastPeerUpdate(conn.Remote()) +} + +func (peer *Peer) handleBroadcastTCP(msg []byte) { + peer.Router.Topology.RebuildRoutes() + peer.ForEachConnection(func(_ PeerName, conn Connection) { + conn.(*LocalConnection).SendTCP(msg) + }) +} + +func (peer *Peer) broadcastPeerUpdate(peers ...*Peer) { + peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer)...))) +} + +func (peer *Peer) checkConnectionLimit() error { + if 0 != peer.Router.ConnLimit && peer.ConnectionCount() >= peer.Router.ConnLimit { + return fmt.Errorf("Connection limit reached (%v)", peer.Router.ConnLimit) + } + return nil +} diff --git a/peer_cache.go b/peer_cache.go new file mode 100644 index 00000000..a6c29f4e --- /dev/null +++ b/peer_cache.go @@ -0,0 +1,311 @@ +package weave + +import ( + "bytes" + "encoding/gob" + "fmt" + "io" + "log" + "sync" +) + +type PeerCache struct { + sync.RWMutex + table map[PeerName]*Peer +} + +func NewPeerCache() *PeerCache { + return &PeerCache{table: make(map[PeerName]*Peer)} +} + +func (cache *PeerCache) FetchWithDefault(peer *Peer) *Peer { + cache.RLock() + res, found := cache.fetchAlias(peer) + cache.RUnlock() + if found { + return res + } + cache.Lock() + defer cache.Unlock() + res, found = cache.fetchAlias(peer) + if found { + return res + } + cache.table[peer.Name] = peer + peer.IncrementLocalRefCount() + return peer +} + +func (cache *PeerCache) Fetch(name PeerName) (*Peer, bool) { + cache.RLock() + defer cache.RUnlock() + peer, found := cache.table[name] + return peer, found // GRRR, why can't I inline this!? +} + +func (cache *PeerCache) ForEach(fun func(PeerName, *Peer)) { + cache.RLock() + defer cache.RUnlock() + for name, peer := range cache.table { + fun(name, peer) + } +} + +func (cache *PeerCache) ApplyUpdate(update []byte, router *Router) ([]byte, error) { + cache.Lock() + + newPeers, decodedUpdate, decodedConns, err := cache.decodeUpdate(update, router) + if err != nil { + cache.Unlock() + return nil, err + } + + // By this point, we know the update doesn't refer to any peers we + // have no knowledge of. We can now apply the update. Start by + // adding in any new peers into the cache. + for name, newPeer := range newPeers { + cache.table[name] = newPeer + } + + // Now apply the updates + newUpdate := cache.applyUpdate(decodedUpdate, decodedConns, router) + + for _, peerRemoved := range cache.garbageCollect(router.Ourself) { + delete(newUpdate, peerRemoved.Name) + } + + // Don't need to hold cache lock any longer + cache.Unlock() + + return encodePeersMap(newUpdate), nil +} + +func (cache *PeerCache) EncodeAllPeers() []byte { + cache.RLock() + defer cache.RUnlock() + return encodePeersMap(cache.table) +} + +func EncodePeers(peers ...*Peer) []byte { + buf := new(bytes.Buffer) + enc := gob.NewEncoder(buf) + for _, peer := range peers { + peer.encodePeer(enc) + } + return buf.Bytes() +} + +func (cache *PeerCache) GarbageCollect(router *Router) []*Peer { + cache.Lock() + defer cache.Unlock() + return cache.garbageCollect(router.Ourself) +} + +func (cache *PeerCache) String() string { + var buf bytes.Buffer + cache.ForEach(func(name PeerName, peer *Peer) { + buf.WriteString(fmt.Sprint(peer, "\n")) + peer.ForEachConnection(func(remoteName PeerName, _ Connection) { + buf.WriteString(fmt.Sprintf(" -> %v\n", remoteName)) + }) + }) + return buf.String() +} + +func (cache *PeerCache) fetchAlias(peer *Peer) (*Peer, bool) { + if existingPeer, found := cache.table[peer.Name]; found { + if existingPeer.UID == peer.UID { + existingPeer.IncrementLocalRefCount() + return existingPeer, true + } else { + return nil, true + } + } + return nil, false +} + +func (cache *PeerCache) garbageCollect(ourself *Peer) []*Peer { + removed := []*Peer{} + for name, peer := range cache.table { + found, _ := ourself.HasPathTo(peer, false) + if !found && !peer.IsLocallyReferenced() { + log.Println("Removing unreachable", peer) + delete(cache.table, name) + ourself.Router.Macs.Delete(peer) + removed = append(removed, peer) + } + } + return removed +} + +func encodePeersMap(peers map[PeerName]*Peer) []byte { + buf := new(bytes.Buffer) + enc := gob.NewEncoder(buf) + for _, peer := range peers { + peer.encodePeer(enc) + } + return buf.Bytes() +} + +func (cache *PeerCache) decodeUpdate(update []byte, router *Router) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]byte, err error) { + newPeers = make(map[PeerName]*Peer) + decodedUpdate = []*Peer{} + decodedConns = [][]byte{} + + updateBuf := new(bytes.Buffer) + updateBuf.Write(update) + decoder := gob.NewDecoder(updateBuf) + + for { + nameByte, uid, version, connsBuf, decErr := decodePeerNoConns(decoder) + if decErr == io.EOF { + break + } else if decErr != nil { + err = decErr + return + } + name := PeerNameFromBin(nameByte) + newPeer := NewPeer(name, uid, version, router) + decodedUpdate = append(decodedUpdate, newPeer) + decodedConns = append(decodedConns, connsBuf) + existingPeer, found := cache.table[name] + if !found { + newPeers[name] = newPeer + } else if existingPeer.UID != newPeer.UID { + err = NameCollisionError{Name: newPeer.Name} + return + } + } + + unknownPeers := false + for _, connsBuf := range decodedConns { + connsIterator(connsBuf, func(remoteNameByte []byte, _ string) { + remoteName := PeerNameFromBin(remoteNameByte) + if _, found := newPeers[remoteName]; found { + return + } + if _, found := cache.table[remoteName]; found { + return + } + // Update refers to a peer which we have no knowledge + // of. Thus we can't apply the update. Abort. + unknownPeers = true + }) + } + if unknownPeers { + err = UnknownPeersError{} + return + } + return +} + +func (cache *PeerCache) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte, router *Router) map[PeerName]*Peer { + newUpdate := make(map[PeerName]*Peer) + for idx, newPeer := range decodedUpdate { + connsBuf := decodedConns[idx] + name := newPeer.Name + // guaranteed to find peer in the cache.table + peer := cache.table[name] + if peer != newPeer { + if peer.Version() > newPeer.Version() { + // we know more about this one than the update. If + // peer is ourself, this is slightly racey (further + // changes could occur to ourself in the mean + // time). But it doesn't matter as we know that we're + // already > newPeer.Version() and that's not going to + // change. + newUpdate[name] = peer + continue + } else if peer == router.Ourself { + // nobody but us updates us + continue + } else if peer.Version() == newPeer.Version() { + // implication is that connections are equal too + continue + } + } + // If we're here, either it was a new peer, or the update has + // more info about the peer than we do. Either case, we need + // to set version and conns and include the updated peer in + // the outgoing update. + + // Can peer have been updated by anyone else in the mean time? + // No - we know that peer is not ourself, so the only prospect + // for an update would be someone else calling + // router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on + // the peers cache, so there can be no race here. + conns := readConnsMap(peer, connsBuf, cache.table) + peer.SetVersionAndConnections(newPeer.Version(), conns) + newUpdate[name] = peer + } + return newUpdate +} + +func (peer *Peer) encodePeer(enc *gob.Encoder) { + peer.RLock() + defer peer.RUnlock() + + checkFatal(enc.Encode(peer.NameByte)) + checkFatal(enc.Encode(peer.UID)) + checkFatal(enc.Encode(peer.version)) + + connsBuf := new(bytes.Buffer) + connsEnc := gob.NewEncoder(connsBuf) + for _, conn := range peer.connections { + // DANGER holding rlock on peer, going to take rlock on conn + if !conn.Established() { + continue + } + checkFatal(connsEnc.Encode(conn.Remote().NameByte)) + checkFatal(connsEnc.Encode(conn.RemoteTCPAddr())) + } + checkFatal(enc.Encode(connsBuf.Bytes())) +} + +func decodePeerNoConns(dec *gob.Decoder) (nameByte []byte, uid uint64, version uint64, conns []byte, err error) { + err = dec.Decode(&nameByte) + if err != nil { + return + } + err = dec.Decode(&uid) + if err != nil { + return + } + err = dec.Decode(&version) + if err != nil { + return + } + err = dec.Decode(&conns) + if err == io.EOF { + err = nil + } + return +} + +func connsIterator(input []byte, fun func([]byte, string)) { + buf := new(bytes.Buffer) + buf.Write(input) + dec := gob.NewDecoder(buf) + for { + var nameByte []byte + err := dec.Decode(&nameByte) + if err == io.EOF { + return + } + checkFatal(err) + var foundAt string + checkFatal(dec.Decode(&foundAt)) + fun(nameByte, string(foundAt)) + } +} + +func readConnsMap(peer *Peer, buf []byte, table map[PeerName]*Peer) map[PeerName]Connection { + conns := make(map[PeerName]Connection) + connsIterator(buf, func(nameByte []byte, remoteTCPAddr string) { + name := PeerNameFromBin(nameByte) + remotePeer := table[name] + conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr) + conns[name] = conn + }) + return conns +} diff --git a/peer_name_hash.go b/peer_name_hash.go new file mode 100644 index 00000000..2ff84b97 --- /dev/null +++ b/peer_name_hash.go @@ -0,0 +1,45 @@ +// +build peer_name_hash + +// Let peer names be sha256 hashes...of anything (as long as it's +// unique). +package weave + +import ( + "crypto/sha256" + "encoding/hex" +) + +type PeerName string + +const ( + PeerNameFlavour = "hash" + NameSize = sha256.Size >> 1 + UnknownPeerName = PeerName("") +) + +func PeerNameFromUserInput(userInput string) (PeerName, error) { + // fixed-length identity + nameByteAry := sha256.Sum256([]byte(userInput)) + return PeerNameFromBin(nameByteAry[:NameSize]), nil +} + +func PeerNameFromString(nameStr string) (PeerName, error) { + if _, err := hex.DecodeString(nameStr); err != nil { + return UnknownPeerName, err + } + return PeerName(nameStr), nil +} + +func PeerNameFromBin(nameByte []byte) PeerName { + return PeerName(hex.EncodeToString(nameByte)) +} + +func (name PeerName) Bin() []byte { + res, err := hex.DecodeString(string(name)) + checkFatal(err) + return res +} + +func (name PeerName) String() string { + return string(name) +} diff --git a/peer_name_mac.go b/peer_name_mac.go new file mode 100644 index 00000000..6d572922 --- /dev/null +++ b/peer_name_mac.go @@ -0,0 +1,57 @@ +// The !peer_name_alternative effectively makes this the default, +// i.e. to choose an alternative, run "go build -tags +// 'peer_name_alternative peer_name_hash'" + +// +build peer_name_mac !peer_name_alternative + +// Let peer names be MACs... +// +// MACs need to be unique across our network, or bad things will +// happen anyway. So they make pretty good candidates for peer +// names. And doing so is pretty efficient both computationally and +// network overhead wise. +// +// Note that we do not mandate *what* MAC should be used as the peer +// name. In particular it doesn't actually have to be the MAC of, say, +// the network interface the peer is sniffing on. +package weave + +import ( + "net" +) + +// PeerName is used as a map key. Since net.HardwareAddr isn't +// suitable for that - it's a slice, and slices can't be map keys - we +// convert that to/from uint64. +type PeerName uint64 + +const ( + PeerNameFlavour = "mac" + NameSize = 6 + UnknownPeerName = PeerName(0) +) + +func PeerNameFromUserInput(userInput string) (PeerName, error) { + res, err := PeerNameFromString(userInput) + return res, err +} + +func PeerNameFromString(nameStr string) (PeerName, error) { + mac, err := net.ParseMAC(nameStr) + if err != nil { + return UnknownPeerName, err + } + return PeerName(macint(mac)), nil +} + +func PeerNameFromBin(nameByte []byte) PeerName { + return PeerName(macint(net.HardwareAddr(nameByte))) +} + +func (name PeerName) Bin() []byte { + return intmac(uint64(name)) +} + +func (name PeerName) String() string { + return intmac(uint64(name)).String() +} diff --git a/router.go b/router.go new file mode 100644 index 00000000..4e555941 --- /dev/null +++ b/router.go @@ -0,0 +1,283 @@ +package weave + +import ( + "bytes" + "code.google.com/p/gopacket/layers" + "code.google.com/p/gopacket/pcap" + "encoding/binary" + "fmt" + "io" + "log" + "net" + "syscall" + "time" +) + +const macMaxAge = 10 * time.Minute + +func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit int, bufSz int, logFrame func(string, []byte, *layers.Ethernet)) *Router { + router := &Router{ + Iface: iface, + Macs: NewMacCache(macMaxAge), + Peers: NewPeerCache(), + ConnLimit: connLimit, + BufSz: bufSz, + LogFrame: logFrame} + if len(password) > 0 { + router.Password = &password + } + ourself := NewPeer(name, 0, 0, router) + router.Ourself = router.Peers.FetchWithDefault(ourself) + router.Ourself.StartLocalPeer() + log.Println("Local identity is", router.Ourself.Name) + + return router +} + +func (router *Router) UsingPassword() bool { + return router.Password != nil +} + +func (router *Router) Start() { + router.ConnectionMaker = StartConnectionMaker(router) + router.Topology = StartTopology(router) + router.UDPListener = router.listenUDP(Port) + router.listenTCP(Port) +} + +func (router *Router) Status() string { + var buf bytes.Buffer + buf.WriteString(fmt.Sprintln("Local identity is", router.Ourself.Name)) + buf.WriteString(fmt.Sprintln("Sniffing traffic on", router.Iface)) + buf.WriteString(fmt.Sprintf("MACs:\n%s", router.Macs)) + buf.WriteString(fmt.Sprintf("Peers:\n%s", router.Peers)) + buf.WriteString(fmt.Sprintf("Topology:\n%s", router.Topology)) + buf.WriteString(fmt.Sprintf("Reconnects:\n%s", router.ConnectionMaker)) + return buf.String() +} + +func (router *Router) Sniff() { + log.Println("Sniffing traffic on", router.Iface) + + handle := router.createPCap(true, 65535, router.BufSz) + defer handle.Close() + + ourself := router.Ourself + dec := NewEthernetDecoder() + + localAddrs, err := router.Iface.Addrs() + checkFatal(err) + checkFrameTooBig := dec.CheckFrameTooBigFunc(router.Iface.HardwareAddr, localAddrs, handle) + + mac := router.Iface.HardwareAddr + if router.Macs.Enter(mac, ourself) { + log.Println("Discovered our MAC", mac) + } + for { + pkt, _, err := handle.ReadPacketData() + checkFatal(err) + router.LogFrame("Sniffed", pkt, nil) + checkWarn(router.handleCapturedPacket(pkt, dec, ourself, checkFrameTooBig)) + } +} + +func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder, ourself *Peer, checkFrameTooBig func(error) error) error { + dec.DecodeLayers(frameData) + decodedLen := len(dec.decoded) + if decodedLen == 0 { + return nil + } + srcMac := dec.eth.SrcMAC + srcPeer, found := router.Macs.Lookup(srcMac) + // We need to filter out frames we injected ourselves. For such + // frames, the srcMAC will have been recorded as associated with a + // different peer. + if found && srcPeer != ourself { + return nil + } + if router.Macs.Enter(srcMac, ourself) { + log.Println("Discovered local MAC", srcMac) + } + if dec.DropFrame() { + return nil + } + dstMac := dec.eth.DstMAC + dstPeer, found := router.Macs.Lookup(dstMac) + if found && dstPeer == ourself { + return nil + } + df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0) + if df { + router.LogFrame("Forwarding DF", frameData, &dec.eth) + } else { + router.LogFrame("Forwarding", frameData, &dec.eth) + } + if !found || dec.BroadcastFrame() { + return checkFrameTooBig(ourself.Broadcast(df, frameData, dec)) + } else { + return checkFrameTooBig(ourself.Forward(dstPeer, df, frameData, dec)) + } +} + +func (router *Router) listenTCP(localPort int) { + localAddr, err := net.ResolveTCPAddr("tcp4", fmt.Sprint(":", localPort)) + checkFatal(err) + ln, err := net.ListenTCP("tcp4", localAddr) + checkFatal(err) + go func() { + defer ln.Close() + for { + tcpConn, err := ln.AcceptTCP() + if err != nil { + log.Println(err) + continue + } + router.acceptTCP(tcpConn) + } + }() +} + +func (router *Router) acceptTCP(tcpConn *net.TCPConn) { + // someone else is dialing us, so our udp sender is the conn + // on Port and we wait for them to send us something on UDP to + // start. + connRemote := NewRemoteConnection(router.Ourself, nil, tcpConn.RemoteAddr().String()) + NewLocalConnection(connRemote, UnknownPeerName, tcpConn, nil, router) +} + +func (router *Router) listenUDP(localPort int) *net.UDPConn { + localAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprint(":", localPort)) + checkFatal(err) + conn, err := net.ListenUDP("udp4", localAddr) + checkFatal(err) + f, err := conn.File() + defer f.Close() + checkFatal(err) + fd := int(f.Fd()) + // This one makes sure all packets we send out do not have DF set on them. + err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DONT) + checkFatal(err) + go router.udpReader(conn) + return conn +} + +func (router *Router) udpReader(conn *net.UDPConn) { + defer conn.Close() + handle := router.createPCap(false, 0, 0) + dec := NewEthernetDecoder() + handleUDPPacket := router.handleUDPPacketFunc(dec, handle) + buf := make([]byte, MaxUDPPacketSize) + for { + n, sender, err := conn.ReadFromUDP(buf) + if err == io.EOF { + return + } else if err != nil { + log.Println("ignoring UDP read error", err) + continue + } else if n < NameSize { + continue // TODO something different? + } else { + name := PeerNameFromBin(buf[:NameSize]) + packet := make([]byte, n-NameSize) + copy(packet, buf[NameSize:n]) + udpPacket := &UDPPacket{ + Name: name, + Packet: packet, + Sender: sender} + peerConn, found := router.Ourself.ConnectionTo(name) + if !found { + continue + } + relayConn, ok := peerConn.(*LocalConnection) + if !ok { + continue + } + checkWarn(relayConn.Decryptor.IterateFrames(handleUDPPacket, udpPacket)) + } + } +} + +func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, handle *pcap.Handle) FrameConsumer { + localAddrs, err := router.Iface.Addrs() + checkFatal(err) + checkFrameTooBig := dec.CheckFrameTooBigFunc(router.Iface.HardwareAddr, localAddrs, handle) + + return func(relayConn *LocalConnection, sender *net.UDPAddr, srcNameByte, dstNameByte []byte, frameLen uint16, frame []byte) error { + srcName := PeerNameFromBin(srcNameByte) + dstName := PeerNameFromBin(dstNameByte) + srcPeer, found := router.Peers.Fetch(srcName) + if !found { + return nil + } + dstPeer, found := router.Peers.Fetch(dstName) + if !found { + return nil + } + + dec.DecodeLayers(frame) + decodedLen := len(dec.decoded) + df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0) + srcMac := dec.eth.SrcMAC + + if dstPeer != router.Ourself { + // it's not for us, we're just relaying it + if decodedLen == 0 { + return nil + } + if router.Macs.Enter(srcMac, srcPeer) { + log.Println("Discovered remote MAC", srcMac, "at", srcPeer.Name) + } + return checkFrameTooBig(router.Ourself.Relay(srcPeer, dstPeer, df, frame, dec)) + } + + if relayConn.Remote().Name == srcPeer.Name { + if frameLen == 0 { + relayConn.SetRemoteUDPAddr(sender) + return nil + } else if frameLen == FragTestSize && bytes.Equal(frame, FragTest) { + relayConn.SendTCP(ProtocolFragmentationReceivedByte) + return nil + } else if frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery) { + return nil + } + } + + if decodedLen == 0 { + return nil + } + + if dec.IsPMTUVerify() && relayConn.Remote().Name == srcPeer.Name { + frameLenBytes := []byte{0, 0} + binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead)) + relayConn.SendTCP(Concat(ProtocolPMTUVerifiedByte, frameLenBytes)) + return nil + } + + if router.Macs.Enter(srcMac, srcPeer) { + log.Println("Discovered remote MAC", srcMac, "at", srcPeer.Name) + } + router.LogFrame("Injecting", frame, &dec.eth) + checkWarn(handle.WritePacketData(frame)) + dstPeer, found = router.Macs.Lookup(dec.eth.DstMAC) + if !found || dec.BroadcastFrame() || dstPeer != router.Ourself { + df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0) + checkFrameTooBig(router.Ourself.RelayBroadcast(srcPeer, df, frame, dec)) + } + return nil + } +} + +func (router *Router) createPCap(promisc bool, snaplen int, buffSize int) *pcap.Handle { + inactive, err := pcap.NewInactiveHandle(router.Iface.Name) + checkFatal(err) + defer inactive.CleanUp() + checkFatal(inactive.SetPromisc(promisc)) + checkFatal(inactive.SetSnapLen(snaplen)) + checkFatal(inactive.SetTimeout(-1)) + checkFatal(inactive.SetImmediateMode(true)) + checkFatal(inactive.SetBufferSize(buffSize)) + active, err := inactive.Activate() + checkFatal(err) + checkFatal(active.SetDirection(pcap.DirectionIn)) + return active +} diff --git a/topology.go b/topology.go new file mode 100644 index 00000000..c0db270c --- /dev/null +++ b/topology.go @@ -0,0 +1,175 @@ +package weave + +import ( + "bytes" + "fmt" + "log" + "sort" +) + +func (topo *Topology) Unicast(name PeerName) (PeerName, bool) { + topo.RLock() + defer topo.RUnlock() + hop, found := topo.unicast[name] + return hop, found +} + +func (topo *Topology) Broadcast(name PeerName) []PeerName { + topo.RLock() + defer topo.RUnlock() + hops, found := topo.broadcast[name] + if !found { + return []PeerName{} + } + return hops +} + +func (topo *Topology) String() string { + var buf bytes.Buffer + topo.RLock() + defer topo.RUnlock() + buf.WriteString(fmt.Sprintln("unicast:")) + for name, hop := range topo.unicast { + buf.WriteString(fmt.Sprintf("%s -> %s\n", name, hop)) + } + buf.WriteString(fmt.Sprintln("broadcast:")) + for name, hops := range topo.broadcast { + buf.WriteString(fmt.Sprintf("%s -> %v\n", name, hops)) + } + return buf.String() +} + +func StartTopology(router *Router) *Topology { + queryChan := make(chan *Interaction, ChannelSize) + state := &Topology{ + router: router, + queryChan: queryChan, + unicast: make(map[PeerName]PeerName), + broadcast: make(map[PeerName][]PeerName)} + state.unicast[router.Ourself.Name] = UnknownPeerName + state.broadcast[router.Ourself.Name] = []PeerName{} + go state.queryLoop(queryChan) + return state +} + +// ACTOR client API + +const ( + TFetchAll = iota + TRebuildRoutes = iota +) + +func (topo *Topology) FetchAll() []byte { + resultChan := make(chan interface{}, 0) + topo.queryChan <- &Interaction{ + code: TFetchAll, + resultChan: resultChan} + result := <-resultChan + return result.([]byte) +} + +// Async. +func (topo *Topology) RebuildRoutes() { + topo.queryChan <- &Interaction{code: TRebuildRoutes} +} + +// ACTOR server + +func (topo *Topology) queryLoop(queryChan <-chan *Interaction) { + for { + query, ok := <-queryChan + if !ok { + return + } + switch { + case query.code == TRebuildRoutes: + unicast := topo.buildUnicastRoutes() + broadcast := topo.buildBroadcastRoutes() + topo.Lock() + topo.unicast = unicast + topo.broadcast = broadcast + topo.Unlock() + case query.code == TFetchAll: + query.resultChan <- topo.router.Peers.EncodeAllPeers() + default: + log.Fatal("Unexpected topology query:", query) + } + } +} + +func (topo *Topology) buildUnicastRoutes() map[PeerName]PeerName { + // He we are calculating all the routes for the question: if *we* + // want to send a packet to Peer X, what is the next hop? + // + // When we sniff a packet, we determine the destination peer + // ourself. Consequently, we can relay the packet via any + // arbitrary peers - the intermediate peers do not have to have + // any knowledge of the MAC address at all. Thus there's no need + // to exchange knowledge of MAC addresses, nor any constraints on + // the routes that we construct. + // + // Because currently we do not have weightings on the connections + // between peers, there is no need to use Dijkstra's + // algorithm. Instead, we can use the simpler and cheaper + // breadth-first widening. + // + // Currently we make no attempt to perform any balancing of any + // sort. That could happen in the future. + + unicast := make(map[PeerName]PeerName) + ourself := topo.router.Ourself + unicast[ourself.Name] = UnknownPeerName + nextWorklist := []*Peer{} + ourself.ForEachConnection(func(remoteName PeerName, conn Connection) { + unicast[remoteName] = remoteName + nextWorklist = append(nextWorklist, conn.Remote()) + }) + for len(nextWorklist) > 0 { + worklist := nextWorklist + sort.Sort(ListOfPeers(worklist)) + nextWorklist = []*Peer{} + for _, peer := range worklist { + localName := peer.Name + peer.ForEachConnection(func(remoteName PeerName, conn Connection) { + if _, found := unicast[remoteName]; found { + return + } + // We now know how to get to remoteName: the same + // way we get to peer. However, we only record this as + // a valid path if remoteName has peer in its + // connections. I.e. it must be a symmetric + // connection. + if _, found := conn.Remote().ConnectionTo(localName); found { + unicast[remoteName] = unicast[localName] + nextWorklist = append(nextWorklist, conn.Remote()) + } + }) + } + } + return unicast +} + +func (topo *Topology) buildBroadcastRoutes() map[PeerName][]PeerName { + // For broadcast, we need to construct which of our connected + // peers we should pass frames on to should we receive any + // broadcast originally from peer X + broadcast := make(map[PeerName][]PeerName) + ourself := topo.router.Ourself + + topo.router.Peers.ForEach(func(name PeerName, peer *Peer) { + hops := []PeerName{} + found, reached := peer.HasPathTo(ourself, true) + if found { + ourself.ForEachConnection(func(remoteName PeerName, conn Connection) { + if _, found := reached[remoteName]; found { + return + } + if _, found := conn.Remote().ConnectionTo(ourself.Name); found { + hops = append(hops, remoteName) + } + }) + } + broadcast[name] = hops + }) + return broadcast +} diff --git a/types.go b/types.go new file mode 100644 index 00000000..f206b48d --- /dev/null +++ b/types.go @@ -0,0 +1,274 @@ +package weave + +import ( + "bytes" + "code.google.com/p/go-bit/bit" + "code.google.com/p/gopacket" + "code.google.com/p/gopacket/layers" + "encoding/gob" + "net" + "sync" + "time" +) + +type Router struct { + Ourself *Peer + Iface *net.Interface + PeersSubscribeChan chan<- chan<- map[string]Peer + Macs *MacCache + Peers *PeerCache + UDPListener *net.UDPConn + Topology *Topology + ConnectionMaker *ConnectionMaker + Password *[]byte + ConnLimit int + BufSz int + LogFrame func(string, []byte, *layers.Ethernet) +} + +type Peer struct { + sync.RWMutex + Name PeerName + NameByte []byte + connections map[PeerName]Connection + version uint64 + UID uint64 + Router *Router + localRefCount uint64 + queryChan chan<- *PeerInteraction +} + +type PeerInteraction struct { + Interaction + payload interface{} +} + +type Connection interface { + Local() *Peer + Remote() *Peer + RemoteTCPAddr() string + Established() bool + Shutdown() +} + +type RemoteConnection struct { + local *Peer + remote *Peer + remoteTCPAddr string +} + +type LocalConnection struct { + sync.RWMutex + RemoteConnection + TCPConn *net.TCPConn + tcpSender TCPSender + remoteUDPAddr *net.UDPAddr + established bool + stackFrag bool + clientPMTU int + pmtuVerified bool + SessionKey *[32]byte + heartbeatStop chan<- interface{} + forwardChan chan<- *ForwardedFrame + forwardChanDF chan<- *ForwardedFrame + stopForward chan<- interface{} + stopForwardDF chan<- interface{} + Decryptor Decryptor + Router *Router + UID uint64 + shutdown bool + queryChan chan<- *ConnectionInteraction +} + +type ConnectionInteraction struct { + Interaction + payload interface{} +} + +type UDPPacket struct { + Name PeerName + Packet []byte + Sender *net.UDPAddr +} + +type EthernetDecoder struct { + eth layers.Ethernet + ip layers.IPv4 + decoded []gopacket.LayerType + parser *gopacket.DecodingLayerParser +} + +type FrameTooBigError struct { + PMTU int + frame *ForwardedFrame +} + +type UnknownPeersError struct { +} + +type NameCollisionError struct { + Name PeerName +} + +type LocalAddress struct { + ip net.IP + network *net.IPNet +} + +type ForwardedFrame struct { + srcPeer *Peer + dstPeer *Peer + frame []byte +} + +type Interaction struct { + code int + resultChan chan<- interface{} +} + +type Topology struct { + sync.RWMutex + queryChan chan<- *Interaction + unicast map[PeerName]PeerName + broadcast map[PeerName][]PeerName + router *Router +} + +type ConnectionMakerInteraction struct { + Interaction + foundAt string + name PeerName +} + +type ConnectionMaker struct { + router *Router + queryChan chan<- *ConnectionMakerInteraction + activeConnections map[string]string + failedConnections map[PeerName]*FailedConnection +} + +type FailedConnection struct { + foundAt map[string]bool + tryAfter time.Time + tryInterval time.Duration + attemptCount int +} + +// UDPSender interface and implementations + +type UDPSender interface { + Send([]byte) error + Shutdown() error +} + +type SimpleUDPSender struct { + conn *LocalConnection + udpConn *net.UDPConn +} + +type RawUDPSender struct { + ipBuf gopacket.SerializeBuffer + opts gopacket.SerializeOptions + udpHeader *layers.UDP + socket *net.IPConn + conn *LocalConnection +} + +// TCPSender interface and implementations + +type TCPSender interface { + Send([]byte) error +} + +type SimpleTCPSender struct { + encoder *gob.Encoder +} + +type EncryptedTCPSender struct { + sync.RWMutex + outerEncoder *gob.Encoder + innerEncoder *gob.Encoder + buffer *bytes.Buffer + conn *LocalConnection + msgCount int +} + +type EncryptedTCPMessage struct { + Number int + Body []byte +} + +// TCPReceiver interface and implementations + +type TCPReceiver interface { + Decode([]byte) ([]byte, error) +} + +type SimpleTCPReceiver struct { +} + +type EncryptedTCPReceiver struct { + conn *LocalConnection + decoder *gob.Decoder + buffer *bytes.Buffer + msgCount int +} + +// Encryptor interface and implementations + +type Encryptor interface { + FrameOverhead() int + PacketOverhead() int + IsEmpty() bool + Bytes() []byte + AppendFrame(*ForwardedFrame) + TotalLen() int +} + +type NonEncryptor struct { + buf []byte + bufTail []byte + buffered int + prefixLen int +} + +type NaClEncryptor struct { + NonEncryptor + buf []byte + offset uint16 + nonce *[24]byte + nonceChan chan *[24]byte + flags uint16 + prefixLen int + conn *LocalConnection + df bool +} + +// Decryptor interface and implementations + +type FrameConsumer func(*LocalConnection, *net.UDPAddr, []byte, []byte, uint16, []byte) error + +type Decryptor interface { + IterateFrames(FrameConsumer, *UDPPacket) error + ReceiveNonce([]byte) + Shutdown() +} + +type NonDecryptor struct { + conn *LocalConnection +} + +type NaClDecryptor struct { + NonDecryptor + instance *NaClDecryptorInstance + instanceDF *NaClDecryptorInstance +} + +type NaClDecryptorInstance struct { + nonce *[24]byte + previousNonce *[24]byte + usedOffsets *bit.Set + previousUsedOffsets *bit.Set + highestOffsetSeen uint16 + nonceChan chan *[24]byte +} diff --git a/utils.go b/utils.go new file mode 100644 index 00000000..23b9193c --- /dev/null +++ b/utils.go @@ -0,0 +1,95 @@ +package weave + +import ( + "crypto/rand" + "fmt" + "log" + "net" +) + +func checkFatal(e error) { + if e != nil { + log.Fatal(e) + } +} + +func checkWarn(e error) { + if e != nil { + log.Println(e) + } +} + +func PosixError(err error) error { + if err == nil { + return nil + } + operr, ok := err.(*net.OpError) + if !ok { + return nil + } + return operr.Err +} + +func (ftbe FrameTooBigError) Error() string { + return fmt.Sprint("Frame too big error. PMTU is ", ftbe.PMTU) +} + +func (upe UnknownPeersError) Error() string { + return fmt.Sprint("Reference to unknown peers") +} + +func (nce NameCollisionError) Error() string { + return fmt.Sprint("Multiple peers found with same name: ", nce.Name) +} + +func (packet UDPPacket) String() string { + return fmt.Sprintf("UDP Packet\n name: %s\n sender: %v\n payload: % X", packet.Name, packet.Sender, packet.Packet) +} + +func Concat(elems ...[]byte) []byte { + res := []byte{} + for _, e := range elems { + res = append(res, e...) + } + return res +} + +func randUint64() (r uint64) { + buf := make([]byte, 8) + _, err := rand.Read(buf) + checkFatal(err) + for _, v := range buf { + r <<= 8 + r |= uint64(v) + } + return +} + +func macint(mac net.HardwareAddr) (r uint64) { + for _, b := range mac { + r <<= 8 + r |= uint64(b) + } + return +} + +func intmac(key uint64) (r net.HardwareAddr) { + r = make([]byte, 6) + for i := 5; i >= 0; i-- { + r[i] = byte(key) + key >>= 8 + } + return +} + +type ListOfPeers []*Peer + +func (lop ListOfPeers) Len() int { + return len(lop) +} +func (lop ListOfPeers) Swap(i, j int) { + lop[i], lop[j] = lop[j], lop[i] +} +func (lop ListOfPeers) Less(i, j int) bool { + return lop[i].Name < lop[i].Name +} diff --git a/weave/Dockerfile b/weave/Dockerfile new file mode 100644 index 00000000..7d08a408 --- /dev/null +++ b/weave/Dockerfile @@ -0,0 +1,6 @@ +FROM scratch +WORKDIR /home/weave +ADD ./weave /home/weave/ +CMD [] +ENTRYPOINT ["/home/weave/weave", "-iface", "ethcr", "-wait"] +EXPOSE 6783/tcp 6783/udp diff --git a/weave/Makefile b/weave/Makefile new file mode 100644 index 00000000..1d1d36a4 --- /dev/null +++ b/weave/Makefile @@ -0,0 +1,22 @@ +.DEFAULT: all + +all: docker-image + +weave: ../*.go main.go + go build -ldflags '-extldflags "-static"' + +.PHONY: docker-image + +docker-image: /tmp/wedo.tar + +/tmp/wedo.tar: Dockerfile weave + sudo docker build -t zettio/weave . + sudo docker save zettio/weave > /tmp/wedo.tar + +pipework: + wget --no-check-certificate -nv https://raw.github.com/jpetazzo/pipework/master/pipework + chmod +x pipework + +clean: + -sudo docker rmi zettio/weave + rm -f weave pipework /tmp/wedo.tar diff --git a/weave/docker-ns b/weave/docker-ns new file mode 100755 index 00000000..1a5a9d2e --- /dev/null +++ b/weave/docker-ns @@ -0,0 +1,36 @@ +#!/bin/bash + +# this borrows from https://github.com/jpetazzo/pipework + +set -e + +GUESTNAME=$1 +shift 1 + +[ "$GUESTNAME" ] || { + echo "Syntax:" + echo "docker-ns ..." + exit 1 +} + +DOCKERPID=$(docker inspect --format='{{ .State.Pid }}' $GUESTNAME) +[ "$DOCKERPID" = 0 ] && { + echo "Docker inspect returned invalid PID 0" + exit 1 +} +[ "$DOCKERPID" = "" ] && { + echo "Container $GUESTNAME unknown to Docker." + exit 1 +} + +NSPID=$DOCKERPID + +[ ! -d /var/run/netns ] && mkdir -p /var/run/netns +[ -f /var/run/netns/$NSPID ] && rm -f /var/run/netns/$NSPID +ln -s /proc/$NSPID/ns/net /var/run/netns/$NSPID + +ip netns exec $NSPID $@ + +[ -f /var/run/netns/$NSPID ] && rm -f /var/run/netns/$NSPID + +exit 0 diff --git a/weave/main.go b/weave/main.go new file mode 100644 index 00000000..95d54f37 --- /dev/null +++ b/weave/main.go @@ -0,0 +1,146 @@ +package main + +import ( + "github.com/zettio/weave" + "code.google.com/p/gopacket/layers" + "crypto/sha256" + "flag" + "fmt" + "github.com/davecheney/profile" + "log" + "net" + "os" + "os/signal" + "runtime" + "syscall" + "time" +) + +func ensureInterface(ifaceName string, wait bool) (iface *net.Interface, err error) { + iface, err = findInterface(ifaceName) + if err == nil || !wait { + return + } + log.Println("Waiting for interface", ifaceName, "to come up") + for err != nil { + time.Sleep(1 * time.Second) + iface, err = findInterface(ifaceName) + } + log.Println("Interface", ifaceName, "is up") + return +} + +func findInterface(ifaceName string) (iface *net.Interface, err error) { + iface, err = net.InterfaceByName(ifaceName) + if err != nil { + return iface, fmt.Errorf("Unable to find interface %s", ifaceName) + } + if 0 == (net.FlagUp & iface.Flags) { + return iface, fmt.Errorf("Interface %s is not up", ifaceName) + } + return +} + +func main() { + + log.SetPrefix(weave.Protocol + " ") + log.SetFlags(log.Ldate | log.Ltime | log.Lmicroseconds) + + runtime.GOMAXPROCS(runtime.NumCPU()) + + var ( + ifaceName string + routerName string + password string + wait bool + debug bool + prof string + peers []string + connLimit int + bufSz int + ) + + flag.StringVar(&ifaceName, "iface", "", "name of interface to read from") + flag.StringVar(&routerName, "name", "", "name of router (defaults to MAC)") + flag.StringVar(&password, "password", "", "network password") + flag.BoolVar(&wait, "wait", false, "wait for interface to be created and come up") + flag.BoolVar(&debug, "debug", false, "enable debug logging") + flag.StringVar(&prof, "profile", "", "enable profiling and write profiles to given path") + flag.IntVar(&connLimit, "connlimit", 10, "connection limit (defaults to 10, set to 0 for unlimited)") + flag.IntVar(&bufSz, "bufsz", 8, "capture buffer size in MB (defaults to 8MB)") + flag.Parse() + peers = flag.Args() + + if ifaceName == "" { + fmt.Println("Missing required parameter 'iface'") + os.Exit(1) + } + iface, err := ensureInterface(ifaceName, wait) + if err != nil { + log.Fatal(err) + } + + if connLimit < 0 { + connLimit = 0 + } + + if routerName == "" { + routerName = iface.HardwareAddr.String() + } + + ourName, err := weave.PeerNameFromUserInput(routerName) + if err != nil { + log.Fatal(err) + } + + var logFrame func(string, []byte, *layers.Ethernet) + if debug { + logFrame = func(prefix string, frame []byte, eth *layers.Ethernet) { + h := fmt.Sprintf("%x", sha256.Sum256(frame)) + if eth == nil { + log.Println(prefix, len(frame), "bytes (", h, ")") + } else { + log.Println(prefix, len(frame), "bytes (", h, "):", eth.SrcMAC, "->", eth.DstMAC) + } + } + } else { + logFrame = func(prefix string, frame []byte, eth *layers.Ethernet) {} + } + + if prof != "" { + p := *profile.CPUProfile + p.ProfilePath = prof + defer profile.Start(&p).Stop() + } + + router := weave.NewRouter(iface, ourName, []byte(password), connLimit, bufSz*1024*1024, logFrame) + router.Start() + startSignalHandlers(router) + for _, peer := range peers { + targetPeer := peer + go func() { + if err := router.Ourself.CreateConnection(targetPeer, weave.UnknownPeerName); err != nil { + log.Println(err) + } + }() + } + router.Sniff() +} + +func startSignalHandlers(router *weave.Router) { + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGQUIT, syscall.SIGUSR1) + buf := make([]byte, 1<<20) + go func() { + for { + sig := <-sigs + switch sig { + case syscall.SIGQUIT: + runtime.Stack(buf, true) + log.Printf("=== received SIGQUIT ===\n*** goroutine dump...\n%s\n*** end\n", buf) + case syscall.SIGUSR1: + log.Printf("=== received SIGUSR1 ===\n*** status...\n%s\n*** end\n", router.Status()) + } + } + }() +} diff --git a/weave/wedo b/weave/wedo new file mode 100755 index 00000000..dc86b645 --- /dev/null +++ b/weave/wedo @@ -0,0 +1,48 @@ +#!/bin/bash +set -e + +BRIDGE=weave +CONTAINER_IFNAME=ethcr +MTU=65535 + +COMMAND=$1 +IPADDR=$2 +shift 2 + +# pre-create bridge so we can set the MTU +[ ! -d /sys/class/net/$BRIDGE ] && { + ip link add dev $BRIDGE type bridge > /dev/null 2>&1 + ip link set dev $BRIDGE address $(od -txC -An -N6 /dev/random | tr \ : | tail -c+2) + # attempting to set the bridge MTU to a high value directly + # fails. Bridges take the lowest MTU of their interfaces. So + # instead we create a temporary interface with the desired MTU, + # attach that to the bridge, and then remove it again. + ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy + ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE + ip link del dev v${CONTAINER_IFNAME}du + ip link set dev $BRIDGE up +} + +# We set the router name to the bridge mac since that is stable across +# re-creations of the containers. +# +# TODO use the mac of one of the physical host interfaces (eth0, +# wlan0, etc) so the name becomes stable across host restarts. +MACADDR=$(&2 + +docker-ns $CONTAINER ethtool -K $IFACE tx off > /dev/null + +echo $CONTAINER