New signed message parsing method. Added SET_KEY message parsing.

Several changes in the interface between transport and protocol layers.
We now correctly add peer info learned from incoming non-HELLO messages because
we need it for sending out PONGs more correctly.
Removed half-implemented "connect" code.

5 files changed, 200 insertions, 151 deletions

diff --git a/src/org/gnu/clojure/gnunet/core.clj b/src/org/gnu/clojure/gnunet/core.clj
new file mode 100644
index 0000000..f3e77e3
--- /dev/null
+++ b/src/org/gnu/clojure/gnunet/core.clj
@@ -0,0 +1,54 @@
+(ns org.gnu.clojure.gnunet.core
+  (:use (org.gnu.clojure.gnunet parser message peer crypto)
+    clojure.contrib.monads))
+
+(def message-type-core-set-key 80)
+(def message-type-core-encrypted-message 81)
+(def message-type-core-ping 82)
+(def message-type-core-pong 83)
+
+(def signature-purpose-set-key 3)
+
+(def parse-set-key
+  (domonad parser-m
+    [sender-status parse-int32
+     signed (parse-signed
+              (domonad parser-m
+                [creation-time parse-date
+                 encrypted-key (items signature-size)
+                 peer-id (items id-size)]
+                {:creation-time creation-time
+                 :encrypted-key encrypted-key
+                 :peer-id peer-id}))
+    :let [signature-purpose (:purpose signed)]
+    :when (= signature-purpose signature-purpose-set-key)
+    signature (items signature-size)]
+    (conj
+      {:sender-status sender-status
+       :signed-material (:signed-material signed)
+       :signature signature}
+      (:parsed signed))))
+
+
+(defn handle-set-key!
+  [peer remote-peer message]
+  (when-let [set-key (first (parse-set-key (:bytes message)))]
+    (cond
+      (not (= (:peer-id set-key) (seq (:id peer)))) (.write *out* "SET_KEY not for me\n")
+      (not (rsa-verify (:public-key remote-peer)
+             (:signed-material set-key)
+             (:signature set-key))) (.write *out* "SET_KEY invalid signature\n")
+      :else (do (.write *out* "Set key message ")
+              (.write *out* (.toString set-key))
+              (.write *out* "\n"))
+      )
+  ))
+
+(defn handle-receive!
+  [peer remote-peer message]
+  (condp = (:message-type message)
+    message-type-core-set-key (handle-set-key! peer remote-peer message)
+    message-type-core-encrypted-message nil
+    message-type-core-ping nil
+    message-type-core-pong nil
+    nil))
diff --git a/src/org/gnu/clojure/gnunet/crypto.clj b/src/org/gnu/clojure/gnunet/crypto.clj
index 43688ee..cc57251 100644
--- a/src/org/gnu/clojure/gnunet/crypto.clj
+++ b/src/org/gnu/clojure/gnunet/crypto.clj
@@ -8,6 +8,33 @@
 
 (def signature-size 256)
 
+(defn encode-signed
+  [purpose inner-material]
+  (concat
+    (encode-int32 (+ 8 (count inner-material)))
+    (encode-int32 purpose)
+    inner-material))
+
+(defn parse-signed
+  "Produces a parser for a signed portion of a GNUnet message given a parser for
+   the contained signed material. The produced parser will fail if the given
+   parser does not successfully consume the entire signed material."
+  [signed-material-parser]
+  (fn
+    [input]
+    (when-let [[[size purpose inner-material] residue]
+               ((domonad parser-m [size parse-uint32
+                                   :when (<= 8 size)
+                                   purpose parse-uint32
+                                   inner-material (items (- size 8))]
+                  [size purpose inner-material]) input)]
+      (when-let [[parsed inner-residue]
+                 (signed-material-parser inner-material)]
+        (if (empty? inner-residue)
+          [{:purpose purpose
+            :signed-material (take size input)
+            :parsed parsed} residue])))))
+
 (defn generate-rsa-keypair
   "Generate a 2048 bit RSA keypair."
   []
diff --git a/src/org/gnu/clojure/gnunet/peer.clj b/src/org/gnu/clojure/gnunet/peer.clj
index 5f2202c..4118449 100644
--- a/src/org/gnu/clojure/gnunet/peer.clj
+++ b/src/org/gnu/clojure/gnunet/peer.clj
@@ -4,25 +4,20 @@
     java.util.concurrent.ConcurrentLinkedQueue
     java.security.SecureRandom))
 
-(defstruct remote-peer
+(defstruct remote-peer-struct
   ;; java.security.PublicKey
   :public-key
   
   ;; 64 byte (512-bit) sequence
   :id
   
-  ;; agent of a map associating transport names (strings) to maps associating
+  ;; agent of a map associating transport names (String) to maps associating
   ;; transport addresses (byte vector) to maps containing {:expiration
   ;; (java.util.Date) :latency (int, if validated)}
-  :transport-addresses-agent
-  
-  ;; agent of a map of {:transport (map from peer:transports-agent, if
-  ;; connection is in progress) :address (single map form of address from
-  ;; :transport-addresses-agent)}
-  :connection-agent)
+  :transport-addresses-agent)
 
-(def peer (apply create-struct (concat
-  (keys (struct-map remote-peer))
+(def peer-struct (apply create-struct (concat
+  (keys (struct-map remote-peer-struct))
   (list
     ;; java.security.PrivateKey
     :private-key
@@ -30,8 +25,7 @@
     ;; agent of a map of peer IDs to struct remote-peer
     :remote-peers-agent
     
-    ;; agent of a map of transport names (String) to maps of {:connect!
-    ;; :emit-message!}
+    ;; agent of a map of transport names (String) to maps of {:emit-message!}
     :transports-agent
     
     ;; java.nio.channels.Selector
@@ -54,7 +48,7 @@
 (defn generate-id
   "Generate the SHA-512 digest of the encoded public key."
   [public-key]
-  (sha-512 (encode-rsa-public-key public-key)))
+  (vec (sha-512 (encode-rsa-public-key public-key))))
 
 (def id-size (count (sha-512 ())))
 
@@ -72,7 +66,7 @@
 (defn new-peer [options]
   (let [selector (Selector/open)
         continuations (ConcurrentLinkedQueue.)]
-    (struct-map peer
+    (struct-map peer-struct
       :public-key (.getPublic (:keypair options))
       :id (generate-id (.getPublic (:keypair options)))
       :transport-addresses-agent (agent {})
diff --git a/src/org/gnu/clojure/gnunet/transport.clj b/src/org/gnu/clojure/gnunet/transport.clj
index 56d499b..e6c60ea 100644
--- a/src/org/gnu/clojure/gnunet/transport.clj
+++ b/src/org/gnu/clojure/gnunet/transport.clj
@@ -1,5 +1,5 @@
 (ns org.gnu.clojure.gnunet.transport
-  (:use (org.gnu.clojure.gnunet parser message hello peer util crypto)
+  (:use (org.gnu.clojure.gnunet parser message hello peer util crypto core)
     clojure.contrib.monads)
   (:import (java.util Date Calendar)))
 
@@ -11,9 +11,16 @@
   []
   (.getTime (doto (Calendar/getInstance) (.add Calendar/HOUR_OF_DAY 1))))
 
+(defn idle-connection-timeout
+  []
+  (.getTime (doto (Calendar/getInstance) (.add Calendar/MINUTE 5))))
+
 (def message-type-ping 32)
 (def message-type-pong 33)
 
+(def signature-purpose-pong-own 1)
+(def signature-purpose-pong-using 2)
+
 (defn encode-ping
   [ping]
   (concat
@@ -34,51 +41,52 @@
      :transport transport
      :encoded-address encoded-address}))
 
-(defn encode-pong
+(defn encode-pong-signed-material
   [pong]
   (let [transport (encode-utf8 (:transport pong))
         address-length (+ (count transport) (count (:encoded-address pong)))]
-    (concat
-      (encode-int32 (:challenge pong))
-      (:signature pong)
-      (encode-int32 (:signature-size pong))
-      (encode-int32 (:signature-purpose pong))
-      (encode-date (:expiration pong))
-      (:peer-id pong)
-      (encode-int32 address-length)
-      transport
-      (:encoded-address pong))))
+    (encode-signed (:signature-purpose pong)
+      (concat
+        (encode-date (:expiration pong))
+        (:peer-id pong)
+        (encode-int32 address-length)
+        transport
+        (:encoded-address pong)))))
 
-(def signature-purpose-pong-own 1)
-(def signature-purpose-pong-using 2)
-(def pong-signature-offset (+ 4 signature-size))
-(def pong-signature-size (+ 4 4 8 id-size 4))
+(defn encode-pong
+  [pong signed-material]
+  (concat
+    (encode-int32 (:challenge pong))
+    (:signature pong)
+    signed-material))
 
 (def parse-pong
-  (domonad parser-m [challenge parse-int32
-                     signature (items signature-size)
-                     signature-size parse-uint32
-                     signature-purpose parse-uint32
-                     expiration parse-date
-                     peer-id (items id-size)
-                     address-length parse-uint32
-                     transport parse-utf8
-                     :when (>= address-length (count (encode-utf8 transport))) 
-                     encoded-address (items
-                                       (- address-length
-                                         (count (encode-utf8 transport))))
-                     :when (= signature-size
-                             (+ pong-signature-size address-length))
-                     residue (none-or-more item)
-                     :when (= 0 (count residue))]
-    {:challenge challenge
-     :signature signature
-     :signature-size signature-size
-     :signature-purpose signature-purpose
-     :expiration expiration
-     :peer-id peer-id
-     :transport transport
-     :encoded-address encoded-address}))
+  (domonad parser-m
+    [challenge parse-int32
+     signature (items signature-size)
+     signed (parse-signed
+              (domonad parser-m
+                [expiration parse-date
+                 peer-id (items id-size)
+                 address-length parse-uint32
+                 transport parse-utf8
+                 :let [transport-length (count (encode-utf8 transport))]
+                 :when (>= address-length transport-length)
+                 encoded-address (items (- address-length transport-length))]
+                {:expiration expiration
+                 :peer-id peer-id
+                 :transport transport
+                 :encoded-address encoded-address}))
+     :let [signature-purpose (:purpose signed)]
+     :when (or (= signature-purpose signature-purpose-pong-own)
+             (= signature-purpose signature-purpose-pong-using))
+     residue (none-or-more item)
+     :when (empty? residue)]
+    (conj {:challenge challenge
+           :signature signature
+           :signature-purpose signature-purpose
+           :signed-material (:signed-material signed)}
+      (:parsed signed))))
 
 (defn list-transport-addresses
   "Generate a list of transport descriptions."
@@ -119,15 +127,6 @@
   [min-expiration addresses-list]
   (filter #(>= 0 (compare min-expiration (:expiration %))) addresses-list))
 
-(defn new-remote-peer-from-hello
-  [peer-id hello]
-  (struct-map remote-peer
-    :public-key (:public-key hello)
-    :id peer-id
-    :transport-addresses-agent (agent (merge-transport-addresses {}
-                                        (:transport-addresses hello)))
-    :connection-agent (agent {})))
-
 (defn hello-for-peer-message
   [peer]
   {:message-type message-type-hello
@@ -145,23 +144,31 @@
                         :transport (:transport address)
                         :encoded-address (:encoded-address address)})})
 
-(defn- update-transport-addresses
+(defn update-transport-addresses
   [addresses new-addresses]
   (merge-transport-addresses {}
     (expire-transport-addresses (Date.)
       (concat (list-transport-addresses addresses) new-addresses))))
 
-(defn- update-remote-peers
+(defn update-remote-peers
   [remote-peers peer-id hello]
-  (let [id (vec peer-id)
-        remote-peer (remote-peers id)]
+  (let [remote-peer (remote-peers peer-id)]
     (if remote-peer
       (do
         (send (:transport-addresses-agent remote-peer)
           update-transport-addresses
           (:transport-addresses hello))
-        remote-peers)
-      (assoc remote-peers id (new-remote-peer-from-hello peer-id hello)))))
+        (if (:public-key remote-peer)
+          remote-peers
+          (assoc remote-peers peer-id
+            (assoc remote-peer :public-key (:public-key hello)))))
+      (assoc remote-peers peer-id
+        (struct-map remote-peer-struct
+          :public-key (:public-key hello)
+          :id peer-id
+          :transport-addresses-agent (agent
+                                       (merge-transport-addresses {}
+                                         (:transport-addresses hello))))))))
 
 (defn verify-transport-address
   [peer remote-peer address]
@@ -206,84 +213,48 @@
   (when-let [hello (first (parse-hello (:bytes message)))]
     (admit-hello! peer hello)))
 
-(defn best-transport
-  [peer remote-peer]
-  (let [addresses (deref (:transport-addressess-agent remote-peer))
-        current-addresses (expire-transport-addresses (Date.)
-                            (list-transport-addresses addresses))
-        transports (deref (:transports-agent peer))
-        usable-addresses (filter #(contains? transports (:transport %))
-                           current-addresses)
-        sorted-addresses (sort-by #(if-let [latency (:latency %)]
-                                     latency
-                                     Integer/MAX_VALUE)
-                           usable-addresses)
-        best (first sorted-addresses)]
-  {:address best
-   :transport (transports (:transport best))}))
-
-(defn connect-to-peer!
-  [peer remote-peer]
-  (send (:connection-agent remote-peer)
-    (fn [connection]
-      (if (contains? connection :transport)
-        (let [{transport :transport address :address} (best-transport peer
-                                                        remote-peer)]
-          (conj {:transport transport :address address}
-            ((:connect! transport) peer remote-peer address)))))))
-
 (defn send-pong-own!
-  [peer sender-id encoded-address ping]
-  (if-let [transport ((deref (:transport-addresses-agent peer))
-                       (:transport ping))]
+  [peer remote-peer ping]
+  (if-let [transport-addresses ((deref (:transport-addresses-agent peer))
+                                 (:transport ping))]
     ;; XXX: Here we're looking for an exact match, gnunet allows transport
     ;; plugins to do inexact matches.
-    (if (contains? transport (:encoded-address ping))
-      (let [expiration (pong-expiration)
-            address-size (+ (count (encode-utf8 (:transport ping)))
-                           (count (:encoded-address ping)))
-            skeliton-pong (encode-pong {:challenge 0
-                               :signature (repeat signature-size (byte 0))
-                               :signature-size (+ pong-signature-size
-                                                 address-size)
-                               :signature-purpose signature-purpose-pong-own
-                               :expiration expiration
-                               :peer-id (:id peer)
-                               :transport (:transport ping)
-                               :encoded-address (:encoded-address ping)})
-            signature (rsa-sign (:private-key peer)
-                        (drop pong-signature-offset skeliton-pong))
-            encoded-pong (encode-pong {:challenge (:challenge ping)
-                               :signature signature
-                               :signature-size (+ pong-signature-size
-                                                 address-size)
-                               :signature-purpose signature-purpose-pong-own
-                               :expiration expiration
-                               :peer-id (:id peer)
-                               :transport (:transport ping)
-                               :encoded-address (:encoded-address ping)})
-            transport ((deref (:transports-agent peer)) (:transport ping))]
-        ;; XXX: gnunet looks for a "reliable" connection for the pong, or it
-        ;; sends a pong to every known address, here we're just sending it back
-        ;; to where the transport said it came from.
-        ((:emit-messages! transport) transport encoded-address
-          [{:message-type message-type-pong :bytes encoded-pong}])))))
+    (if (contains? transport-addresses (:encoded-address ping))
+      (let [pong {:challenge (:challenge ping)
+                  :signature-purpose signature-purpose-pong-own
+                  :expiration (pong-expiration)
+                  :peer-id (:id peer)
+                  :transport (:transport ping)
+                  :encoded-address (:encoded-address ping)}
+            signed-material (encode-pong-signed-material pong) 
+            signature (rsa-sign (:private-key peer) signed-material)
+            pong (assoc pong :signature signature)
+            encoded-pong (encode-pong pong signed-material)]
+        ;; XXX: gnunet looks for a "reliable" connection for the pong before
+        ;; sending to every known address.
+        (doseq [transport-addresses (deref
+                                      (:transport-addresses-agent remote-peer))
+                address (val transport-addresses)]
+          (if-let [transport ((deref (:transports-agent peer))
+                               (key transport-addresses))]
+            ((:emit-messages! transport) transport (key address)
+              [{:message-type message-type-pong :bytes encoded-pong}])))))))
 
 (defn send-pong-using!
-  [peer sender-id encoded-address ping]
+  [peer remote-peer ping]
   (.write *out* "We don't handle PONG_USING yet!\n")
   )
 
 (defn handle-ping!
-  [peer sender-id encoded-address message]
+  [peer remote-peer message]
   (when-let [ping (first (parse-ping (:bytes message)))]
     (cond
       (not (= (:peer-id ping) (seq (:id peer)))) nil
-      (:transport ping) (send-pong-own! peer sender-id encoded-address ping)
-      :else (send-pong-using! peer sender-id encoded-address ping))))
+      (:transport ping) (send-pong-own! peer remote-peer ping)
+      :else (send-pong-using! peer remote-peer ping))))
 
 (defn check-pending-validation
-  [addresses remote-peer pong encoded-pong]
+  [addresses remote-peer pong]
   (if-let [transport (addresses (:transport pong))]
     (if-let [address (transport (:encoded-address pong))]
       (cond
@@ -291,7 +262,7 @@
           addresses
         (= signature-purpose-pong-own (:signature-purpose pong))
           (if (rsa-verify (:public-key remote-peer)
-                (drop pong-signature-offset encoded-pong)
+                (:signed-material pong)
                 (:signature pong))
             (assoc addresses (:transport pong)
               (assoc transport (:encoded-address pong)
@@ -313,20 +284,26 @@
       (when-let [remote-peer ((deref (:remote-peers-agent peer))
                                (:peer-id pong))]
         (send (:transport-addresses-agent remote-peer) check-pending-validation
-          remote-peer pong (:bytes message))))))
+          remote-peer pong)))))
 
 (defn admit-message!
-  [peer sender-id encoded-address message]
+  [peer sender-id address message]
   (let [string-builder (StringBuilder. "Received message type ")]
       (.append string-builder (:message-type message))
       (.append string-builder " from ")
-      (.append string-builder encoded-address)
+      (.append string-builder (vec (:encoded-address address)))
       (.append string-builder " id ")
       (.append string-builder sender-id)
       (.append string-builder "\n")
       (.write *out* (.toString string-builder)))
-  (condp = (:message-type message)
-    message-type-hello (handle-hello! peer message)
-    message-type-ping (handle-ping! peer sender-id encoded-address message)
-    message-type-pong (handle-pong! peer message)
-    nil))
+  (send (:remote-peers-agent peer)
+    (fn [remote-peers]
+      (let [remote-peers (update-remote-peers remote-peers
+                           sender-id {:transport-addresses [address]})
+            remote-peer (remote-peers sender-id)]
+        (condp = (:message-type message)
+          message-type-hello (handle-hello! peer message)
+          message-type-ping (handle-ping! peer remote-peer message)
+          message-type-pong (handle-pong! peer message)
+          (handle-receive! peer remote-peer message))
+        remote-peers))))
diff --git a/src/org/gnu/clojure/gnunet/udp.clj b/src/org/gnu/clojure/gnunet/udp.clj
index c443e45..6d0c4eb 100644
--- a/src/org/gnu/clojure/gnunet/udp.clj
+++ b/src/org/gnu/clojure/gnunet/udp.clj
@@ -68,14 +68,16 @@
   [peer datagram-channel]
   (let [byte-buffer (doto (ByteBuffer/allocate max-udp-packet-length) (.clear))
         source-address (.receive datagram-channel byte-buffer)
-        encoded-address (encode-address source-address)]
+        address {:transport "udp"
+                 :encoded-address (encode-address source-address)
+                 :expiration (idle-connection-timeout)}]
     (.flip byte-buffer)
     (when-let [{udp :message} (first ((parse-message-types
                                         {message-type-udp parse-udp})
                                        (buffer-seq! byte-buffer)))]
-      (if (not (= (:sender-id udp) (seq (:id peer))))
+      (if (not (= (:sender-id udp) (:id peer)))
         (doseq [message (:messages udp)]
-          (admit-message! peer (:sender-id udp) encoded-address message))))))
+          (admit-message! peer (:sender-id udp) address message))))))
 
 (defn handle-channel-selected!
   [peer datagram-channel selection-key]
@@ -84,10 +86,6 @@
   (if (.isWritable selection-key)
     (handle-channel-writable! peer datagram-channel)))
 
-(defn connect-udp!
-  [peer remote-peer address]
-  {:message-queue (ConcurrentLinkedQueue.)})
-
 (defn- register-datagram-channel!
   [peer port]
   (let [datagram-channel (DatagramChannel/open)
@@ -102,8 +100,7 @@
       (send (:transports-agent peer)
         (fn [transports]
           (assoc transports "udp"
-            {:connect! (partial connect-udp! peer)
-             :emit-messages! (partial emit-messages-udp! peer)
+            {:emit-messages! (partial emit-messages-udp! peer)
              :socket socket
              :selection-key selection-key
              :send-queue (ConcurrentLinkedQueue.)}))))))