Sending/receiving datagrams

I2PAddr.go

@@ -18,18 +18,17 @@ var (
 // The public and private keys associated with an I2P destination. I2P hides the
 // details of exactly what this is, so treat them as blobs, but generally: One
 // pair of DSA keys, one pair of ElGamal keys, and sometimes (almost never) also
-// a certificate. Addr() and Priv() returns you the full content of I2PKeys, 
+// a certificate. String() returns you the full content of I2PKeys and Addr() 
-// which you could/should store to disk if you want to be able to use the I2P 
+// returns the public keys.
-// destination again in the future.
 type I2PKeys struct {
-	addr    I2PAddr
+	addr    I2PAddr       // only the public key
-	priv    string
+	both    string        // both public and private keys
 }
 
-// Creates I2PKeys from an I2PAddr and a private key. The I2PAddr obviously must
+// Creates I2PKeys from an I2PAddr and a public/private keypair string (as 
-// be the public key belonging to the private key. Performs no error checking.
+// generated by String().)
-func NewKeys(addr I2PAddr, priv string) I2PKeys {
+func NewKeys(addr I2PAddr, both string) I2PKeys {
-	return I2PKeys{addr, priv}
+	return I2PKeys{addr, both}
 }
 
 // Returns the public keys of the I2PKeys.
@@ -37,9 +36,10 @@ func (k I2PKeys) Addr() I2PAddr {
 	return k.addr
 }
 
-// Returns the private keys, in I2Ps base64 format.
+// Returns the keys (both public and private), in I2Ps base64 format. Use this
-func (k I2PKeys) Priv() string {
+// when you create sessions.
-	return k.priv
+func (k I2PKeys) String() string {
+	return k.both
 }
 
 // I2PAddr represents an I2P destination, almost equivalent to an IP address.

README.md

@@ -15,11 +15,13 @@ This library is much better than ccondom (that use BOB), much more stable and mu
 * Streaming
     * DialI2P() - Connecting to stuff in I2P
     * Listen()/Accept() - Handling incomming connections
+    * Implements net.Conn and net.Listener
+* Datagrams
+    * Implements net.PacketConn
 
 **Does not work:**
 
-* Datagram sockets
+* Raw packets
-* Raw sockets
 
 ## Documentation ##
 

datagram.go

@@ -1,8 +1,10 @@
 package sam3
 
 import (
+	"bytes"
 	"errors"
 	"net"
+	"strconv"
 	"time"
 )
 
@@ -11,51 +13,118 @@ import (
 // also end-to-end encrypted, signed and includes replay-protection. And they 
 // are also built to be surveillance-resistant (yey!).
 type DatagramSession struct {
-	keys        I2PKeys
+    samAddr     string             // address to the sam bridge (ipv4:port)
-	lport       int
+	id          string             // tunnel name
+	conn        net.Conn           // connection to sam bridge
+	udpconn     *net.UDPConn        // used to deliver datagrams
+	keys        I2PKeys            // i2p destination keys
+	rUDPAddr    *net.UDPAddr        // the SAM bridge UDP-port
 }
 
-// Creates a new datagram session.
+// Creates a new datagram session. udpPort is the UDP port SAM is listening on, 
-func (sam *SAM) NewDatagramSession(tunnelName string, keys I2PKeys, options []string) (*DatagramSession, error) {
+// and if you set it to zero, it will use SAMs standard UDP port.
-	return nil, errors.New("Not implemented.")
+func (s *SAM) NewDatagramSession(id string, keys I2PKeys, options []string, udpPort int) (*DatagramSession, error) {
+	if udpPort > 65335 || udpPort < 0 {
+		return nil, errors.New("udpPort needs to be in the intervall 0-65335")
+	}
+	if udpPort == 0 {
+		udpPort = 7655
+	}
+	lhost, _, err := net.SplitHostPort(s.conn.LocalAddr().String())
+	if err != nil {
+		s.Close()
+		return nil, err
+	}
+	lUDPAddr, err := net.ResolveUDPAddr("udp4", lhost + ":0")
+	if err != nil {
+		return nil, err
+	}
+	udpconn, err := net.ListenUDP("udp4", lUDPAddr)
+	if err != nil {
+		return nil, err
+	}
+	rhost, _, err := net.SplitHostPort(s.conn.RemoteAddr().String())
+	if err != nil {
+		s.Close()
+		return nil, err
+	}
+	rUDPAddr, err := net.ResolveUDPAddr("udp4", rhost + ":" + strconv.Itoa(udpPort))
+	if err != nil {
+		return nil, err
+	}
+	_, lport, err := net.SplitHostPort(udpconn.LocalAddr().String())
+	conn, err := s.newGenericSession("DATAGRAM", id, keys, options, []string{"PORT=" + lport})
+	if err != nil {
+		return nil, err
+	}
+	return &DatagramSession{s.address, id, conn, udpconn, keys, rUDPAddr}, nil
 }
 
 // Reads one datagram sent to the destination of the DatagramSession. Returns 
 // the number of bytes read, from what address it was sent, or an error.
 func (s *DatagramSession) ReadFrom(b []byte) (n int, addr I2PAddr, err error) {
-	return 0, I2PAddr(""), errors.New("Not implemented.")
+	// extra bytes to read the remote address of incomming datagram
+	buf := make([]byte, len(b) + 4096)
+	
+	n, _, err = s.udpconn.ReadFrom(buf)
+	if err != nil {
+		return 0, I2PAddr(""), err
+	}
+	i := bytes.IndexByte(buf, byte('\n'))
+	if i > 4096 || i > n {
+		return 0, I2PAddr(""), errors.New("Could not parse incomming message remote address.")
+	}
+	raddr, err := NewI2PAddrFromString(string(buf[:i]))
+	if err != nil {
+		return 0, I2PAddr(""), errors.New("Could not parse incomming message remote address: " + err.Error())
+	}
+	// shift out the incomming address to contain only the data received
+	copy(b, buf[i+1:i+1+len(b)])
+
+	if ( n - i+1 ) > len(b) {
+		return n-(i+1), raddr, errors.New("Datagram did not fit into your buffer.")
+	}
+	return n-(i+1), raddr, nil
 }
 
 // Sends one signed datagram to the destination specified. At the time of 
 // writing, maximum size is 31 kilobyte, but this may change in the future.
 // Implements net.PacketConn.
 func (s *DatagramSession) WriteTo(b []byte, addr I2PAddr) (n int, err error) {
-	return 0, errors.New("Not implemented.")
+	header := []byte("3.0 " + s.id + " " + addr.String() + "\n")
+	msg := append(header, b...)
+	n, err = s.udpconn.WriteToUDP(msg, s.rUDPAddr)
+	return n, err
 }
 
 // Closes the DatagramSession. Implements net.PacketConn
 func (s *DatagramSession) Close() error {
-	return errors.New("Not implemented.")
+	err := s.conn.Close()
+	err2 := s.udpconn.Close()
+	if err != nil {
+		return err
+	}
+	return err2
 }
 
 // Returns the I2P destination of the DatagramSession. Implements net.PacketConn
-func (s *DatagramSession) LocalAddr() net.Addr {
+func (s *DatagramSession) LocalAddr() I2PAddr {
 	return s.keys.Addr()
 }
 
 // Implements net.PacketConn
 func (s *DatagramSession) SetDeadline(t time.Time) error {
-	return errors.New("Not implemented.")
+	return s.udpconn.SetDeadline(t)
 }
 
 // Implements net.PacketConn
 func (s *DatagramSession) SetReadDeadline(t time.Time) error {
-	return errors.New("Not implemented.")
+	return s.udpconn.SetReadDeadline(t)
 }
 
 // Implements net.PacketConn
 func (s *DatagramSession) SetWriteDeadline(t time.Time) error {
-	return errors.New("Not implemented.")
+	return s.udpconn.SetWriteDeadline(t)
 }
 
 

sam3.go

@@ -122,10 +122,11 @@ func (sam *SAM) Lookup(name string) (I2PAddr, error) {
 
 // Creates a new session with the style of either "STREAM", "DATAGRAM" or "RAW", 
 // for a new I2P tunnel with name id, using the cypher keys specified, with the
-// I2CP/streaminglib-options as specified. Returns the connection used to 
+// I2CP/streaminglib-options as specified. Extra arguments can be specified by
-// control the SAMv3 bridge. The SAM-object should be treated as destroyed after
+// setting extra to something else than []string{}. Returns the connection used
-// calling this function on it.
+// to control the SAMv3 bridge. The SAM-object should be treated as destroyed 
-func (sam *SAM) newGenericSession(style, id string, keys I2PKeys, options []string) (net.Conn, error) {
+// after calling this function on it.
+func (sam *SAM) newGenericSession(style, id string, keys I2PKeys, options []string, extras []string) (net.Conn, error) {
 	sam2, err := NewSAM(sam.address)
 	if err != nil {
 		return nil, errors.New("Unable to create new streaming tunnel.")
@@ -136,7 +137,7 @@ func (sam *SAM) newGenericSession(style, id string, keys I2PKeys, options []stri
 	}
 	
 	conn := sam2.conn
-	scmsg := []byte("SESSION CREATE STYLE=" + style + " ID=" + id + " DESTINATION=" + keys.String() + " " + optStr + "\n")
+	scmsg := []byte("SESSION CREATE STYLE=" + style + " ID=" + id + " DESTINATION=" + keys.String() + " " + optStr + strings.Join(extras, " ") + "\n")
 	for m, i:=0, 0; m!=len(scmsg); i++ {
 		if i == 15 {
 			conn.Close()

sam_test.go

@@ -4,9 +4,10 @@ package sam3
 
 import (
 	"fmt"
+	"runtime"
 	"strings"
 	"testing"
-//	"time"
+	"time"
 )
 
 
@@ -32,7 +33,7 @@ func Test_Basic(t *testing.T) {
 		t.Fail()
 	} else {
 		fmt.Println("\tAddress created: " + keys.Addr().Base32())
-		fmt.Println("\tI2PKeys: " + string(keys.priv)[:50] + "(...etc)")
+		fmt.Println("\tI2PKeys: " + string(keys.both)[:50] + "(...etc)")
 	}
 	
 	addr2, err := sam.Lookup("zzz.i2p")
@@ -122,6 +123,7 @@ func Test_StreamingDial(t *testing.T) {
 		t.Fail()
 		return
 	}
+	fmt.Println("\tNotice: This may fail if your I2P node is not well integrated in the I2P network.")
 	fmt.Println("\tLooking up forum.i2p")
 	forumAddr, err := sam.Lookup("forum.i2p")
 	if err != nil {
@@ -156,6 +158,10 @@ func Test_StreamingServerClient(t *testing.T) {
 	if testing.Short() {
 		return
 	}
+	
+	ncpu := runtime.NumCPU()
+	runtime.GOMAXPROCS(ncpu + 1)
+	
 	fmt.Println("Test_StreamingServerClient")
 	sam, err := NewSAM(yoursam)
 	if err != nil {
@@ -217,9 +223,16 @@ func Test_StreamingServerClient(t *testing.T) {
 		w <- false
 		return
 	}
+	defer l.Close()
 	w <- true
 	fmt.Println("\tServer: Accept()ing on tunnel")
 	conn, err := l.Accept()
+	if err != nil {
+		t.Fail()
+		fmt.Println("Failed to Accept(): " + err.Error())
+		return
+	}
+	defer conn.Close()
 	buf := make([]byte, 512)
 	n,err := conn.Read(buf)
 	fmt.Printf("\tClient exited successfully: %t\n", <-c)
@@ -228,3 +241,84 @@ func Test_StreamingServerClient(t *testing.T) {
 
 
 
+func Test_DatagramServerClient(t *testing.T) {
+//	if testing.Short() {
+//		return
+//	}
+
+	ncpu := runtime.NumCPU()
+	runtime.GOMAXPROCS(ncpu + 1)
+
+	fmt.Println("Test_DatagramServerClient")
+	sam, err := NewSAM(yoursam)
+	if err != nil {
+		t.Fail()
+		return
+	}
+	defer sam.Close()
+	keys, err := sam.NewKeys()
+	if err != nil {
+		t.Fail()
+		return
+	}
+//	fmt.Println("\tServer: My address: " + keys.Addr().Base32())
+	fmt.Println("\tServer: Creating tunnel")
+	ds, err := sam.NewDatagramSession("DGserverTun", keys, []string{"inbound.length=0", "outbound.length=0", "inbound.lengthVariance=0", "outbound.lengthVariance=0", "inbound.quantity=1", "outbound.quantity=1"}, 0)
+	if err != nil {
+		fmt.Println("Server: Failed to create tunnel: " + err.Error())
+		t.Fail()
+		return
+	}
+	c, w := make(chan bool), make(chan bool)
+	go func(c, w chan(bool)) { 
+		sam2, err := NewSAM(yoursam)
+		if err != nil {
+			c <- false
+			return
+		}
+		defer sam2.Close()
+		keys, err := sam2.NewKeys()
+		if err != nil {
+			c <- false
+			return
+		}
+		fmt.Println("\tClient: Creating tunnel")
+		ds2, err := sam2.NewDatagramSession("DGclientTun", keys, []string{"inbound.length=0", "outbound.length=0", "inbound.lengthVariance=0", "outbound.lengthVariance=0", "inbound.quantity=1", "outbound.quantity=1"}, 0)
+		if err != nil {
+			c <- false
+			return
+		}
+		defer ds2.Close()
+//		fmt.Println("\tClient: Servers address: " + ds.LocalAddr().Base32())
+//		fmt.Println("\tClient: Clients address: " + ds2.LocalAddr().Base32())
+		fmt.Println("\tClient: Tries to send datagram to server")
+		for {
+			select {
+				default :
+					_, err = ds2.WriteTo([]byte("Hello datagram-world! <3 <3 <3 <3 <3 <3"), ds.LocalAddr())
+					if err != nil {
+						fmt.Println("\tClient: Failed to send datagram: " + err.Error())
+						c <- false
+						return
+					}
+					time.Sleep(5 * time.Second)
+				case <-w :
+					fmt.Println("\tClient: Sent datagram, quitting.")
+					return
+			}
+		}
+		c <- true
+	}(c, w)
+	buf := make([]byte, 512)
+	fmt.Println("\tServer: ReadFrom() waiting...")
+	n, _, err := ds.ReadFrom(buf)
+	w <- true
+	if err != nil {
+		fmt.Println("\tServer: Failed to ReadFrom(): " + err.Error())
+		t.Fail()
+		return
+	}
+	fmt.Println("\tServer: Received datagram: " + string(buf[:n]))
+//	fmt.Println("\tServer: Senders address was: " + saddr.Base32())
+}
+

stream.go

@@ -7,7 +7,6 @@ import (
 	"net"
 	"strconv"
 	"strings"
-	"sync"
 )
 
 // Represents a streaming session.
@@ -16,9 +15,6 @@ type StreamSession struct {
 	id          string             // tunnel name
 	conn        net.Conn           // connection to sam bridge
 	keys        I2PKeys            // i2p destination keys
- 	listener    *StreamListener    // used for accepting inbound calls
-	l           sync.Mutex         // lock for this struct
-	err         error
 }
 
 // Returns the local tunnel name of the I2P tunnel used for the stream session
@@ -39,11 +35,11 @@ func (ss StreamSession) Keys() I2PKeys {
 // Creates a new StreamSession with the I2CP- and streaminglib options as 
 // specified. See the I2P documentation for a full list of options.
 func (sam *SAM) NewStreamSession(id string, keys I2PKeys, options []string) (*StreamSession, error) {
-	conn, err := sam.newGenericSession("STREAM", id, keys, options)
+	conn, err := sam.newGenericSession("STREAM", id, keys, options, []string{})
 	if err != nil {
 		return nil, err
 	}
-	return &StreamSession{sam.address, id, conn, keys, nil, sync.Mutex{}, nil}, nil
+	return &StreamSession{sam.address, id, conn, keys}, nil
 }
 
 // Dials to an I2P destination and returns a SAMConn, which implements a net.Conn.
@@ -96,7 +92,12 @@ func (s *StreamSession) Listen() (*StreamListener, error) {
 	if err != nil {
 		return nil, err
 	}
-	listener, err := net.Listen("tcp4", ":0")
+	lhost, _, err := net.SplitHostPort(s.conn.LocalAddr().String())
+	if err != nil {
+		sam.Close()
+		return nil, err
+	}
+	listener, err := net.Listen("tcp4", lhost + ":0")
 	_, lport, err := net.SplitHostPort(listener.Addr().String())
 	if err != nil {
 		sam.Close()