Merge lp:~jderose/dmedia/peer-sync into lp:dmedia

[Jason Gerard DeRose]

When a ~/.local/share/dmedia/library.json file is present, UserCouch is configured to listen on 0.0.0.0 and use OAuth, and we advertise this over Avahi.

As noted in the bug report, this feature provides NO PRIVACY WHATSOEVER.

[lp:~jderose/dmedia/peer-sync]

373. By Jason Gerard DeRose on 2012-04-28

Removed a debugging print

[James Raymond (jamesmr)]

review: Approve

=== modified file 'dmedia-service'
--- dmedia-service	2012-04-23 18:30:42 +0000
+++ dmedia-service	2012-04-28 20:46:19 +0000
@@ -31,6 +31,7 @@
 import optparse
 import logging
 import json
+from os import path
 
 import dbus
 import dbus.service
@@ -43,6 +44,7 @@
 from dmedia.core import Core, start_file_server
 from dmedia.service.udisks import UDisks
 from dmedia.service.avahi import Avahi
+from dmedia.service.replicator import Replicator
 
 
 BUS = dmedia.BUS
@@ -56,12 +58,20 @@
 
 def dumps(obj):
     return json.dumps(obj, sort_keys=True, separators=(',', ': '), indent=4)
+    
+    
+def get_config(f):
+    try:
+        return json.load(open(f, 'r'))
+    except Exception:
+        pass
 
 
 class Service(dbus.service.Object):
     usercouch = None
     httpd = None
     avahi = None
+    replicator = None
 
     def __init__(self, bus, env_s):
         self.bus = bus
@@ -74,8 +84,14 @@
     def start_core(self):
         if self.env_s is None:
             self.usercouch = usercouch.UserCouch(dmedia.get_dmedia_dir())
-            env = self.usercouch.bootstrap()
+            f = path.join(self.usercouch.basedir, 'library.json')
+            self.config = get_config(f)
+            if self.config is None:
+                env = self.usercouch.bootstrap()
+            else:
+                env = self.usercouch.bootstrap2(self.config['tokens'])
         else:
+            self.config = None
             env = json.loads(self.env_s)
         self.udisks = UDisks()
         self.core = Core(env, self.udisks.get_parentdir_info)
@@ -91,6 +107,9 @@
         (self.httpd, self.port) = start_file_server(self.core.env)
         self.avahi = Avahi(self.core.env, self.port)
         self.avahi.run()
+        if self.config is not None:
+            self.replicator = Replicator(self.core.env, self.config)
+            self.replicator.run()
 
     def run(self):
         self.start_core()
@@ -119,6 +138,8 @@
             self.usercouch.kill()
         if self.avahi is not None:
             self.avahi.free()
+        if self.replicator is not None:
+            self.replicator.free()
         if self.httpd is not None:
             self.httpd.terminate()
             self.httpd.join()
 
=== modified file 'dmedia/service/replicator.py'
--- dmedia/service/replicator.py	2012-04-24 11:22:28 +0000
+++ dmedia/service/replicator.py	2012-04-28 20:46:19 +0000
@@ -25,6 +25,7 @@
 
 import logging
 import json
+from collections import namedtuple
 
 from microfiber import Server, Database, PreconditionFailed
 import dbus
@@ -32,16 +33,19 @@
 
 log = logging.getLogger()
 system = dbus.SystemBus()
-
+Peer = namedtuple('Peer', 'url names')
 SERVICE = '_usercouch._tcp'
 
 
-def get_body(source, target):
-    return {
+def get_body(source, target, cancel=False):
+    body = {
         'source': source,
         'target': target,
         'continuous': True,
     }
+    if cancel:
+        body['cancel'] = True
+    return body
 
 
 def get_peer(url, dbname, tokens):
@@ -56,11 +60,10 @@
 class Replicator:
     def __init__(self, env, config):
         self.group = None
+        self.env = env
         self.server = Server(env)
         self.library_id = config['library_id']
         self.base_id = self.library_id + '-'
-        self.machine_id = env['machine_id']
-        self.id = self.base_id + self.machine_id
         self.port = env['port']
         self.tokens = config['tokens']
         self.peers = {}
@@ -69,6 +72,8 @@
         self.free()
 
     def run(self):
+        self.machine_id = self.env['machine_id']
+        self.id = self.base_id + self.machine_id
         self.avahi = system.get_object('org.freedesktop.Avahi', '/')
         self.group = system.get_object(
             'org.freedesktop.Avahi',
@@ -106,44 +111,70 @@
         if self.group is not None:
             self.group.Reset(dbus_interface='org.freedesktop.Avahi.EntryGroup')
 
-    def on_ItemNew(self, interface, protocol, name, _type, domain, flags):
-        if name == self.id:  # Ignore what we publish ourselves
+    def on_ItemNew(self, interface, protocol, key, _type, domain, flags):
+        if key == self.id:  # Ignore what we publish ourselves
             return
-        if not name.startswith(self.base_id):  # Ignore other libraries
+        if not key.startswith(self.base_id):  # Ignore other libraries
             return
         (ip, port) = self.avahi.ResolveService(
-            interface, protocol, name, _type, domain, -1, 0,
+            interface, protocol, key, _type, domain, -1, 0,
             dbus_interface='org.freedesktop.Avahi.Server'
         )[7:9]
         url = 'http://{}:{}/'.format(ip, port)
-        log.info('Replicator: new peer %s at %s', name, url)
-        self.peers[name] = url
-        self.replicate(url, 'foo') 
-
-    def on_ItemRemove(self, interface, protocol, name, _type, domain, flags):
-        log.info('Replicator: removing peer %s', name)
-        try:
-            del self.peers[name]
-        except KeyError:
-            pass
-
-    def replicate(self, url, dbname):
-        # Create local DB if needed
-        try:
-            self.server.put(None, dbname)
-        except PreconditionFailed:
-            pass
-
-        # Create remote DB if needed
-        env = {'url': url, 'oauth': self.tokens}
-        db = Database(dbname, env)
-        db.ensure() 
-
-        peer = get_peer(url, dbname, self.tokens)
-        local_to_remote = get_body(dbname, peer)
-        remote_to_local = get_body(peer, dbname)
-        for obj in (local_to_remote, remote_to_local):
-            self.server.post(obj, '_replicate')
+        log.info('Replicator: new peer %s at %s', key, url)
+        self.cancel_all(key)
+        self.peers[key] = Peer(url, [])
+        self.replicate_all(key)
+
+    def on_ItemRemove(self, interface, protocol, key, _type, domain, flags):
+        log.info('Replicator: peer removed %s', key)
+        self.cancel_all(str(key))
+
+    def cancel_all(self, key):
+        p = self.peers.pop(key, None)
+        if p is None:
+            return
+        log.info('Canceling replications for %r', key)
+        for name in p.names:
+            self.replicate(p.url, name, cancel=True)
+
+    def replicate_all(self, key):
+        p = self.peers[key]
+        env = {'url': p.url, 'oauth': self.tokens}
+        remote = Server(env)
+        for name in self.server.get('_all_dbs'):
+            if name.startswith('_'):
+                continue
+            if not (name.startswith('dmedia-0') or name.startswith('novacut-0')):
+                continue
+            # Create remote DB if needed
+            try:
+                remote.put(None, name)
+            except PreconditionFailed:
+                pass
+
+            # Start replication
+            p.names.append(name)
+            self.replicate(p.url, name)
+
+    def replicate(self, url, name, cancel=False):
+        """
+        Start or cancel push replication of database *name* to peer at *url*.
+
+        Security note: we only do push replication because pull replication
+        would allow unauthorized peers to write to our databases via their
+        changes feed.  For both push and pull, there is currently no privacy
+        whatsoever... everything is in cleartext and uses oauth 1.0a. But push
+        replication is the only way to at least prevent malicious data
+        corruption.
+        """
+        if cancel:
+            log.info('Canceling push of %r to %r', name, url)
+        else:
+            log.info('Starting push of %r to %r', name, url)
+        peer = get_peer(url, name, self.tokens)
+        push = get_body(name, peer, cancel)
+        self.server.post(push, '_replicate')
         
         
         
 
=== added file 'init-library'
--- init-library	1970-01-01 00:00:00 +0000
+++ init-library	2012-04-28 20:46:19 +0000
@@ -0,0 +1,25 @@
+#!/usr/bin/python3
+
+import json
+from os import path
+
+from microfiber import random_id
+from usercouch import random_oauth
+
+import dmedia
+
+
+def dumps(obj):
+    return json.dumps(obj, sort_keys=True, separators=(',', ': '), indent=4)
+
+
+obj = {
+    'library_id': random_id(),
+    'tokens': random_oauth(),
+}
+obj_s = dumps(obj)
+
+f = path.join(dmedia.get_dmedia_dir(), 'library.json')
+open(f, 'w').write(obj_s)
+print(obj_s)
+