Push Publishing

mc/nat.go

@@ -119,7 +119,7 @@ func NATConfigFromString(str string) (cfg NATConfig, err error) {
 }
 
 func GetPublicIP() (string, error) {
-	res, err := http.Get("http://ifconfig.me/ip")
+	res, err := http.Get("http://ifconfig.co/ip")
 	if err != nil {
 		return "", err
 	}

mc/query/query.go

@@ -32,6 +32,10 @@ func (q *Query) IsSimpleSelect(sel string) bool {
 	return false
 }
 
+func (q *Query) WithSimpleSelect(sel string) *Query {
+	return &Query{q.Op, q.namespace, SimpleSelector(sel), q.criteria, q.order, q.limit}
+}
+
 type QuerySelector interface {
 	selectorType() string
 }

mcnode/api.go

@@ -360,7 +360,7 @@ func (node *Node) httpRemoteQuery(w http.ResponseWriter, r *http.Request) {
 // POST /merge/{peerId}
 // DATA: MCQL SELECT query
 // Queries a remote peer and merges the resulting statements into the local
-// db; returns the number of statements merged
+// db; returns the number of statements and objects merged
 func (node *Node) httpMerge(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	peerId := vars["peerId"]
@@ -410,6 +410,63 @@ func (node *Node) httpMerge(w http.ResponseWriter, r *http.Request) {
 	fmt.Fprintln(w, ocount)
 }
 
+// POST /push/{peerId}
+// DATA: MCQL SELECT query
+// Pushes statements matching the query to peerId for merge; must be
+// authorized for push by the peer
+// returns the number of statements and objects merged
+func (node *Node) httpPush(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	peerId := vars["peerId"]
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		log.Printf("http/push: Error reading request body: %s", err.Error())
+		return
+	}
+
+	q := string(body)
+
+	qq, err := mcq.ParseQuery(q)
+	if err != nil {
+		apiError(w, http.StatusBadRequest, err)
+		return
+	}
+
+	if !qq.IsSimpleSelect("*") {
+		apiError(w, http.StatusBadRequest, BadQuery)
+		return
+	}
+
+	pid, err := p2p_peer.IDB58Decode(peerId)
+	if err != nil {
+		apiError(w, http.StatusBadRequest, err)
+		return
+	}
+
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
+
+	count, ocount, err := node.doPush(ctx, pid, qq)
+	if err != nil {
+		apiError(w, http.StatusInternalServerError, err)
+		if count > 0 {
+			fmt.Fprintf(w, "Partial push: %d statements merged\n", count)
+		}
+		if ocount > 0 {
+			fmt.Fprintf(w, "Partial push: %d objects merged\n", ocount)
+		}
+		if count < 0 {
+			fmt.Fprintf(w, "Incomplete push: some statements may have been merged\n")
+		}
+
+		return
+	}
+
+	fmt.Fprintln(w, count)
+	fmt.Fprintln(w, ocount)
+}
+
 // POST /delete
 // DATA: MCQL DELTE query
 // Deletes statements from the statement db
@@ -719,6 +776,74 @@ func (node *Node) httpConfigInfoSet(w http.ResponseWriter, r *http.Request) {
 	fmt.Fprintln(w, "OK")
 }
 
+// GET /auth
+// retrieves all peer authorization rules in json
+func (node *Node) httpAuth(w http.ResponseWriter, r *http.Request) {
+	rules := node.auth.toJSON()
+
+	err := json.NewEncoder(w).Encode(rules)
+	if err != nil {
+		log.Printf("Error writing response body: %s", err.Error())
+	}
+}
+
+// GET  /auth/{peerId}
+// POST /auth/{peerId}
+// gets/sets auth rules for peerId
+// rules are specified as a comma separated list of namespaces (or ns wildcards)
+func (node *Node) httpAuthPeer(w http.ResponseWriter, r *http.Request) {
+	apiConfigMethod(w, r, node.httpAuthPeerGet, node.httpAuthPeerSet)
+}
+
+func (node *Node) httpAuthPeerGet(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	peerId := vars["peerId"]
+
+	pid, err := p2p_peer.IDB58Decode(peerId)
+	if err != nil {
+		apiError(w, http.StatusBadRequest, err)
+		return
+	}
+
+	rules := node.auth.getRules(pid)
+	if len(rules) > 0 {
+		fmt.Fprintln(w, strings.Join(rules, ","))
+	}
+}
+
+func (node *Node) httpAuthPeerSet(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	peerId := vars["peerId"]
+
+	pid, err := p2p_peer.IDB58Decode(peerId)
+	if err != nil {
+		apiError(w, http.StatusBadRequest, err)
+		return
+	}
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		log.Printf("http/auth: Error reading request body: %s", err.Error())
+		return
+	}
+
+	rbody := strings.TrimSpace(string(body))
+	if rbody == "" {
+		node.auth.clearRules(pid)
+	} else {
+		rules := strings.Split(rbody, ",")
+		node.auth.setRules(pid, rules)
+	}
+
+	err = node.saveConfig()
+	if err != nil {
+		apiError(w, http.StatusInternalServerError, err)
+		return
+	}
+
+	fmt.Fprintln(w, "OK")
+}
+
 // POST /shutdown
 // shutdown the node
 func (node *Node) httpShutdown(w http.ResponseWriter, r *http.Request) {

mcnode/main.go

@@ -79,6 +79,7 @@ func main() {
 	router.HandleFunc("/query", node.httpQuery)
 	router.HandleFunc("/query/{peerId}", node.httpRemoteQuery)
 	router.HandleFunc("/merge/{peerId}", node.httpMerge)
+	router.HandleFunc("/push/{peerId}", node.httpPush)
 	router.HandleFunc("/delete", node.httpDelete)
 	router.HandleFunc("/data/put", node.httpPutData)
 	router.HandleFunc("/data/get/{objectId}", node.httpGetData)
@@ -88,6 +89,8 @@ func main() {
 	router.HandleFunc("/config/dir", node.httpConfigDir)
 	router.HandleFunc("/config/nat", node.httpConfigNAT)
 	router.HandleFunc("/config/info", node.httpConfigInfo)
+	router.HandleFunc("/auth", node.httpAuth)
+	router.HandleFunc("/auth/{peerId}", node.httpAuthPeer)
 	router.HandleFunc("/dir/list", node.httpDirList)
 	router.HandleFunc("/net/addr", node.httpNetAddr)
 	router.HandleFunc("/shutdown", node.httpShutdown)

mcnode/node.go

@@ -8,6 +8,7 @@ import (
 	ggproto "github.com/gogo/protobuf/proto"
 	p2p_crypto "github.com/libp2p/go-libp2p-crypto"
 	p2p_host "github.com/libp2p/go-libp2p-host"
+	p2p_peer "github.com/libp2p/go-libp2p-peer"
 	p2p_pstore "github.com/libp2p/go-libp2p-peerstore"
 	mc "github.com/mediachain/concat/mc"
 	mcq "github.com/mediachain/concat/mc/query"
@@ -18,6 +19,7 @@ import (
 	"log"
 	"os"
 	"path"
+	"strings"
 	"sync"
 	"time"
 )
@@ -36,6 +38,7 @@ type Node struct {
 	home      string
 	db        StatementDB
 	ds        Datastore
+	auth      PeerAuth
 	mx        sync.Mutex
 	counter   int
 }
@@ -65,6 +68,11 @@ type Datastore interface {
 	Close()
 }
 
+type PeerAuth struct {
+	peers map[p2p_peer.ID][]string
+	mx    sync.Mutex
+}
+
 type NodeInfo struct {
 	Peer      string `json:"peer"`
 	Publisher string `json:"publisher"`
@@ -85,6 +93,9 @@ var (
 	MissingData      = errors.New("Missing statement metadata")
 	UnexpectedData   = errors.New("Unexpected data object")
 	BadData          = errors.New("Bad data object; hash mismatch")
+	BadPush          = errors.New("Bad push value; unexpected object")
+	BadResponse      = errors.New("Bad response; unexpected object")
+	BadRuleset       = errors.New("Bad auth ruleset; unexpected object")
 )
 
 const (
@@ -95,6 +106,12 @@ const (
 
 var statusString = []string{"offline", "online", "public"}
 
+type PushError string
+
+func (s PushError) Error() string {
+	return string(s)
+}
+
 type StreamError struct {
 	Err string `json:"error"`
 }
@@ -256,9 +273,10 @@ func (node *Node) openDS() error {
 
 // persistent configuration
 type NodeConfig struct {
-	Info string `json:"info,omitempty"`
-	NAT  string `json:"nat,omitempty"`
-	Dir  string `json:"dir,omitempty"`
+	Info string                 `json:"info,omitempty"`
+	NAT  string                 `json:"nat,omitempty"`
+	Dir  string                 `json:"dir,omitempty"`
+	Auth map[string]interface{} `json:"auth,omitempty"`
 }
 
 func (node *Node) saveConfig() error {
@@ -268,6 +286,7 @@ func (node *Node) saveConfig() error {
 	if node.dir != nil {
 		cfg.Dir = mc.FormatHandle(*node.dir)
 	}
+	cfg.Auth = node.auth.toJSON()
 
 	bytes, err := json.Marshal(cfg)
 	if err != nil {
@@ -311,7 +330,8 @@ func (node *Node) loadConfig() error {
 		node.dir = &pinfo
 	}
 
-	return nil
+	err = node.auth.fromJSON(cfg.Auth)
+	return err
 }
 
 func (node *Node) doShutdown() {
@@ -323,3 +343,106 @@ func (node *Node) doShutdown() {
 	node.ds.Close()
 	os.Exit(0)
 }
+
+func (auth *PeerAuth) fromJSON(rmap map[string]interface{}) error {
+	auth.mx.Lock()
+	defer auth.mx.Unlock()
+
+	auth.peers = make(map[p2p_peer.ID][]string)
+	for id58, xrules := range rmap {
+		pid, err := p2p_peer.IDB58Decode(id58)
+		if err != nil {
+			return err
+		}
+
+		xxrules, ok := xrules.([]interface{})
+		if !ok {
+			return BadRuleset
+		}
+
+		rules := make([]string, len(xxrules))
+		for x, xxrule := range xxrules {
+			rule, ok := xxrule.(string)
+			if !ok {
+				return BadRuleset
+			}
+
+			rules[x] = rule
+		}
+
+		auth.peers[pid] = rules
+	}
+
+	return nil
+}
+
+func (auth *PeerAuth) toJSON() map[string]interface{} {
+	auth.mx.Lock()
+	defer auth.mx.Unlock()
+
+	rmap := make(map[string]interface{})
+	for pid, rules := range auth.peers {
+		rmap[pid.Pretty()] = rules
+	}
+
+	return rmap
+}
+
+func (auth *PeerAuth) authorize(pid p2p_peer.ID, nss []string) bool {
+	if len(nss) == 0 {
+		return false
+	}
+
+	auth.mx.Lock()
+	defer auth.mx.Unlock()
+
+	rules := auth.peers[pid]
+	if len(rules) == 0 {
+		return false
+	}
+
+	for _, ns := range nss {
+		if !auth.authorizeAllow(rules, ns) {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (auth *PeerAuth) authorizeAllow(rules []string, ns string) bool {
+	for _, rule := range rules {
+		switch {
+		case rule == "*":
+			return true
+
+		case strings.HasSuffix(rule, ".*"):
+			if strings.HasPrefix(ns, rule[:len(rule)-2]) {
+				return true
+			}
+
+		case rule == ns:
+			return true
+		}
+	}
+
+	return false
+}
+
+func (auth *PeerAuth) getRules(pid p2p_peer.ID) []string {
+	auth.mx.Lock()
+	defer auth.mx.Unlock()
+	return auth.peers[pid]
+}
+
+func (auth *PeerAuth) setRules(pid p2p_peer.ID, rules []string) {
+	auth.mx.Lock()
+	auth.peers[pid] = rules
+	auth.mx.Unlock()
+}
+
+func (auth *PeerAuth) clearRules(pid p2p_peer.ID) {
+	auth.mx.Lock()
+	delete(auth.peers, pid)
+	auth.mx.Unlock()
+}