Merge lp:~sergiusens/account-polld/refresh_accounts into lp:~phablet-team/account-polld/trunk

[Sergio Schvezov]

Add support for handling token expiry

This is part of a series of commits from https://code.launchpad.net/~jamesh/account-polld/account-refresh/+merge/227759
I've taken a subset from it which would be good to have now and leave the other part for review.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:22
http://jenkins.qa.ubuntu.com/job/account-polld-ci/11/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/account-polld-utopic-amd64-ci/11
    SUCCESS: http://jenkins.qa.ubuntu.com/job/account-polld-utopic-armhf-ci/11
        deb: http://jenkins.qa.ubuntu.com/job/account-polld-utopic-armhf-ci/11/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/account-polld-utopic-i386-ci/11

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/account-polld-ci/11/rebuild

review: Approve (continuous-integration)

=== modified file 'accounts/account-watcher.c'
--- accounts/account-watcher.c	2014-07-11 12:01:42 +0000
+++ accounts/account-watcher.c	2014-07-24 20:44:55 +0000
@@ -82,7 +82,7 @@
     g_free(info);
 }
 
-static void account_info_notify(AccountInfo *info) {
+static void account_info_notify(AccountInfo *info, GError *error) {
     AgService *service = ag_account_service_get_service(info->account_service);
     const char *service_name = ag_service_get_name(service);
     char *client_id = NULL;
@@ -108,6 +108,7 @@
     info->watcher->callback(info->watcher,
                             info->account_id,
                             service_name,
+                            error,
                             info->enabled,
                             client_id,
                             client_secret,
@@ -124,32 +125,18 @@
 
     GError *error = NULL;
     info->session_data = signon_auth_session_process_finish(session, result, &error);
+    account_info_notify(info, error);
 
     if (error != NULL) {
         fprintf(stderr, "Authentication failed: %s\n", error->message);
         g_error_free(error);
-        return;
     }
-    account_info_notify(info);
 }
 
-static void account_info_enabled_cb(
-    AgAccountService *account_service, gboolean enabled, AccountInfo *info) {
-    fprintf(stderr, "account_info_enabled_cb for %u, enabled=%d\n", info->account_id, enabled);
-    if (info->enabled == enabled) {
-        /* no change */
-        return;
-    }
-    info->enabled = enabled;
-
+static void account_info_login(AccountInfo *info) {
     account_info_clear_login(info);
-    if (!enabled) {
-        // Send notification that account has been disabled */
-        account_info_notify(info);
-        return;
-    }
 
-    AgAuthData *auth_data = ag_account_service_get_auth_data(account_service);
+    AgAuthData *auth_data = ag_account_service_get_auth_data(info->account_service);
     GError *error = NULL;
     fprintf(stderr, "Starting authentication session for account %u\n", info->account_id);
     info->session = signon_auth_session_new(
@@ -157,21 +144,15 @@
         ag_auth_data_get_method(auth_data), &error);
     if (error != NULL) {
         fprintf(stderr, "Could not set up auth session: %s\n", error->message);
+        account_info_notify(info, error);
         g_error_free(error);
         g_object_unref(auth_data);
         return;
     }
 
-    GVariantBuilder builder;
-    g_variant_builder_init(&builder, G_VARIANT_TYPE_VARDICT);
-    g_variant_builder_add(
-        &builder, "{sv}",
-        SIGNON_SESSION_DATA_UI_POLICY,
-        g_variant_new_int32(SIGNON_POLICY_NO_USER_INTERACTION));
-
     info->auth_params = g_variant_ref_sink(
         ag_auth_data_get_login_parameters(
-            auth_data, g_variant_builder_end(&builder)));
+            auth_data, NULL));
 
     signon_auth_session_process_async(
         info->session,
@@ -182,6 +163,24 @@
     ag_auth_data_unref(auth_data);
 }
 
+static void account_info_enabled_cb(
+    AgAccountService *account_service, gboolean enabled, AccountInfo *info) {
+    fprintf(stderr, "account_info_enabled_cb for %u, enabled=%d\n", info->account_id, enabled);
+    if (info->enabled == enabled) {
+        /* no change */
+        return;
+    }
+    info->enabled = enabled;
+
+    if (enabled) {
+        account_info_login(info);
+    } else {
+        account_info_clear_login(info);
+        // Send notification that account has been disabled */
+        account_info_notify(info, NULL);
+    }
+}
+
 static AccountInfo *account_info_new(AccountWatcher *watcher, AgAccountService *account_service) {
     AccountInfo *info = g_new0(AccountInfo, 1);
     info->watcher = watcher;
@@ -258,7 +257,7 @@
     }
     ag_manager_list_free(enabled_accounts);
 
-    return FALSE;
+    return G_SOURCE_REMOVE;
 }
 
 AccountWatcher *account_watcher_new(GHashTable *services_to_watch,
@@ -277,3 +276,28 @@
     g_idle_add(account_watcher_setup, watcher);
     return watcher;
 }
+
+struct refresh_info {
+    AccountWatcher *watcher;
+    AgAccountId account_id;
+};
+
+static gboolean account_watcher_refresh_cb(void *user_data) {
+    struct refresh_info *data = (struct refresh_info *)user_data;
+
+    AccountInfo *info = g_hash_table_lookup(
+        data->watcher->services, GUINT_TO_POINTER(data->account_id));
+    if (info != NULL) {
+        account_info_login(info);
+    }
+
+    return G_SOURCE_REMOVE;
+}
+
+void account_watcher_refresh(AccountWatcher *watcher, unsigned int account_id) {
+    struct refresh_info *data = g_new(struct refresh_info, 1);
+    data->watcher = watcher;
+    data->account_id = account_id;
+    g_idle_add_full(G_PRIORITY_DEFAULT_IDLE, account_watcher_refresh_cb,
+                    data, g_free);
+}

=== modified file 'accounts/account-watcher.h'
--- accounts/account-watcher.h	2014-07-11 12:01:42 +0000
+++ accounts/account-watcher.h	2014-07-24 20:44:55 +0000
@@ -24,6 +24,7 @@
 typedef void (*AccountEnabledCallback)(AccountWatcher *watcher,
                                        unsigned int account_id,
                                        const char *service_name,
+                                       GError *error,
                                        int enabled,
                                        const char *client_id,
                                        const char *client_secret,
@@ -35,4 +36,6 @@
                                     AccountEnabledCallback callback,
                                     void *user_data);
 
+void account_watcher_refresh(AccountWatcher *watcher, unsigned int account_id);
+
 #endif

=== modified file 'accounts/accounts.c'
--- accounts/accounts.c	2014-07-11 13:44:14 +0000
+++ accounts/accounts.c	2014-07-24 20:44:55 +0000
@@ -1,5 +1,4 @@
 #include "_cgo_export.h"
-#include "account-watcher.h"
 
 AccountWatcher *watch_for_services(void *array_of_service_names, int length) {
     /* Transfer service names to hash table */
@@ -11,8 +10,24 @@
         g_hash_table_insert(services_to_watch, g_strdup(service_names[i].p), NULL);
     }
 
+    if (FALSE) {
+        /* The Go callback doesn't quite match the
+         * AccountEnabledCallback function prototype, so we cast the
+         * argument in the account_watcher_new() call below.
+         *
+         * This is just a check to see that the function still has the
+         * prototype we expect.
+         */
+        void (*unused)(void *watcher,
+                       unsigned int account_id, char *service_name,
+                       GError *error, int enabled,
+                       char *client_id, char *client_secret,
+                       char *access_token, char *token_secret,
+                       void *user_data) = authCallback;
+    }
+
     AccountWatcher *watcher = account_watcher_new(
-        services_to_watch, authCallback, NULL);
+        services_to_watch, (AccountEnabledCallback)authCallback, NULL);
     g_hash_table_unref(services_to_watch);
     return watcher;
 }

=== modified file 'accounts/accounts.go'
--- accounts/accounts.go	2014-07-11 12:14:35 +0000
+++ accounts/accounts.go	2014-07-24 20:44:55 +0000
@@ -21,17 +21,26 @@
 #cgo pkg-config: glib-2.0 libaccounts-glib libsignon-glib
 #include <stdlib.h>
 #include <glib.h>
-
-typedef struct _AccountWatcher AccountWatcher;
+#include "account-watcher.h"
 
 AccountWatcher *watch_for_services(void *array_of_service_names, int length);
 */
 import "C"
-import "unsafe"
+import (
+	"errors"
+	"sync"
+	"unsafe"
+)
+
+type Watcher struct {
+	C       <-chan AuthData
+	watcher *C.AccountWatcher
+}
 
 type AuthData struct {
 	AccountId   uint
 	ServiceName string
+	Error       error
 	Enabled     bool
 
 	ClientId     string
@@ -41,28 +50,47 @@
 }
 
 var (
-	mainLoop     *C.GMainLoop
+	mainLoopOnce sync.Once
 	authChannels = make(map[*C.AccountWatcher]chan<- AuthData)
+	authChannelsLock sync.Mutex
 )
 
-func WatchForService(serviceNames... string) <-chan AuthData {
-	if mainLoop == nil {
-		mainLoop = C.g_main_loop_new(nil, C.gboolean(1))
+func startMainLoop() {
+	mainLoopOnce.Do(func() {
+		mainLoop := C.g_main_loop_new(nil, C.gboolean(1))
 		go C.g_main_loop_run(mainLoop)
-	}
+	})
+}
 
-	watcher := C.watch_for_services(unsafe.Pointer(&serviceNames[0]), C.int(len(serviceNames)))
+// NewWatcher creates a new account watcher for the given service names
+func NewWatcher(serviceNames... string) *Watcher {
+	w := new(Watcher)
+	w.watcher = C.watch_for_services(unsafe.Pointer(&serviceNames[0]), C.int(len(serviceNames)))
 
 	ch := make(chan AuthData)
-	authChannels[watcher] = ch
-	return ch
+	w.C = ch
+	authChannelsLock.Lock()
+	authChannels[w.watcher] = ch
+	authChannelsLock.Unlock()
+
+	startMainLoop()
+
+	return w
+}
+
+// Refresh requests that the token for the given account be refreshed.
+// The new access token will be delivered over the watcher's channel.
+func (w *Watcher) Refresh(accountId uint) {
+	C.account_watcher_refresh(w.watcher, C.uint(accountId))
 }
 
 //export authCallback
-func authCallback(watcher unsafe.Pointer, accountId C.uint, serviceName *C.char, enabled C.int, clientId, clientSecret, accessToken, tokenSecret *C.char, userData unsafe.Pointer) {
+func authCallback(watcher unsafe.Pointer, accountId C.uint, serviceName *C.char, error *C.GError, enabled C.int, clientId, clientSecret, accessToken, tokenSecret *C.char, userData unsafe.Pointer) {
 	// Ideally the first argument would be of type
 	// *C.AccountWatcher, but that fails with Go 1.2.
+	authChannelsLock.Lock()
 	ch := authChannels[(*C.AccountWatcher)(watcher)]
+	authChannelsLock.Unlock()
 	if ch == nil {
 		// Log the error
 		return
@@ -71,6 +99,9 @@
 	var data AuthData
 	data.AccountId = uint(accountId)
 	data.ServiceName = C.GoString(serviceName)
+	if error != nil {
+		data.Error = errors.New(C.GoString((*C.char)(error.message)))
+	}
 	if enabled != 0 {
 		data.Enabled = true
 	}

=== modified file 'cmd/account-polld/account_manager.go'
--- cmd/account-polld/account_manager.go	2014-07-17 18:35:36 +0000
+++ cmd/account-polld/account_manager.go	2014-07-24 20:44:55 +0000
@@ -19,7 +19,6 @@
 
 import (
 	"log"
-	"sync"
 	"time"
 
 	"launchpad.net/account-polld/accounts"
@@ -27,59 +26,74 @@
 )
 
 type AccountManager struct {
+	watcher   *accounts.Watcher
 	authData  accounts.AuthData
-	authMutex *sync.Mutex
 	plugin    plugins.Plugin
 	interval  time.Duration
 	postWatch chan *PostWatch
-	terminate chan bool
+	authChan  chan accounts.AuthData
 }
 
 const DEFAULT_INTERVAL = time.Duration(60 * time.Second)
 
-func NewAccountManager(authData accounts.AuthData, postWatch chan *PostWatch, plugin plugins.Plugin) *AccountManager {
+func NewAccountManager(watcher *accounts.Watcher, postWatch chan *PostWatch, plugin plugins.Plugin) *AccountManager {
 	return &AccountManager{
+		watcher:   watcher,
 		plugin:    plugin,
-		authData:  authData,
-		authMutex: &sync.Mutex{},
+		interval:  DEFAULT_INTERVAL,
 		postWatch: postWatch,
-		interval:  DEFAULT_INTERVAL,
-		terminate: make(chan bool),
+		authChan:  make(chan accounts.AuthData, 1),
 	}
 }
 
 func (a *AccountManager) Delete() {
-	a.terminate <- true
+	close(a.authChan)
 }
 
 func (a *AccountManager) Loop() {
-	defer close(a.terminate)
+	var ok bool
+	if a.authData, ok = <- a.authChan; !ok {
+		return
+	}
 L:
 	for {
 		log.Println("Polling set to", a.interval, "for", a.authData.AccountId)
 		select {
 		case <-time.After(a.interval):
 			a.poll()
-		case <-a.terminate:
-			break L
+		case a.authData, ok = <-a.authChan:
+			if !ok {
+				break L
+			}
+			a.poll()
 		}
 	}
 	log.Printf("Ending poll loop for account %d", a.authData.AccountId)
 }
 
 func (a *AccountManager) poll() {
-	a.authMutex.Lock()
-	defer a.authMutex.Unlock()
-
 	if !a.authData.Enabled {
 		log.Println("Account", a.authData.AccountId, "no longer enabled")
 		return
 	}
 
+	if a.authData.Error != nil {
+		log.Println("Account", a.authData.AccountId, "failed to authenticate:", a.authData.Error)
+		return
+	}
+
 	if n, err := a.plugin.Poll(&a.authData); err != nil {
 		log.Print("Error while polling ", a.authData.AccountId, ": ", err)
 		// penalizing the next poll
 		a.interval += DEFAULT_INTERVAL
+
+		// If the error indicates that the authentication
+		// token has expired, request reauthentication and
+		// mark data as disabled.
+		if err == plugins.ErrTokenExpired {
+			a.watcher.Refresh(a.authData.AccountId)
+			a.authData.Enabled = false
+		}
 	} else if len(n) > 0 {
 		// on success we reset the timeout to the default interval
 		a.interval = DEFAULT_INTERVAL
@@ -88,7 +102,5 @@
 }
 
 func (a *AccountManager) updateAuthData(authData accounts.AuthData) {
-	a.authMutex.Lock()
-	defer a.authMutex.Unlock()
-	a.authData = authData
+	a.authChan <- authData
 }

=== modified file 'cmd/account-polld/main.go'
--- cmd/account-polld/main.go	2014-07-18 11:42:52 +0000
+++ cmd/account-polld/main.go	2014-07-24 20:44:55 +0000
@@ -69,9 +69,10 @@
 }
 
 func monitorAccounts(postWatch chan *PostWatch) {
+	watcher := accounts.NewWatcher(SERVICENAME_GMAIL, SERVICENAME_FACEBOOK, SERVICENAME_TWITTER)
 	mgr := make(map[uint]*AccountManager)
 L:
-	for data := range accounts.WatchForService(SERVICENAME_GMAIL, SERVICENAME_FACEBOOK, SERVICENAME_TWITTER) {
+	for data := range watcher.C {
 		if account, ok := mgr[data.AccountId]; ok {
 			if data.Enabled {
 				log.Printf("New account data for %d - was %#v, now is %#v", data.AccountId, account.authData, data)
@@ -98,7 +99,8 @@
 				log.Println("Unhandled account with id", data.AccountId, "for", data.ServiceName)
 				continue L
 			}
-			mgr[data.AccountId] = NewAccountManager(data, postWatch, plugin)
+			mgr[data.AccountId] = NewAccountManager(watcher, postWatch, plugin)
+			mgr[data.AccountId].updateAuthData(data)
 			go mgr[data.AccountId].Loop()
 		}
 	}

=== modified file 'cmd/account-watcher-test/main.go'
--- cmd/account-watcher-test/main.go	2014-07-11 12:07:47 +0000
+++ cmd/account-watcher-test/main.go	2014-07-24 20:44:55 +0000
@@ -9,7 +9,7 @@
 
 func main() {
 	// Expects a list of service names as command line arguments
-	for data := range accounts.WatchForService(os.Args[1:]...) {
+	for data := range accounts.NewWatcher(os.Args[1:]...).C {
 		fmt.Printf("%#v\n", data)
 	}
 }

=== modified file 'plugins/facebook/facebook.go'
--- plugins/facebook/facebook.go	2014-07-17 18:35:36 +0000
+++ plugins/facebook/facebook.go	2014-07-24 20:44:55 +0000
@@ -62,6 +62,9 @@
 		if err := decoder.Decode(&result); err != nil {
 			return nil, err
 		}
+		if result.Error.Code == 190 {
+			return nil, plugins.ErrTokenExpired
+		}
 		return nil, &result.Error
 	}
 

=== modified file 'plugins/facebook/facebook_test.go'
--- plugins/facebook/facebook_test.go	2014-07-18 11:38:36 +0000
+++ plugins/facebook/facebook_test.go	2014-07-24 20:44:55 +0000
@@ -23,6 +23,8 @@
 	"testing"
 
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/account-polld/plugins"
 )
 
 type S struct{}
@@ -48,10 +50,18 @@
 	errorBody = `
 {
   "error": {
-    "message": "Message describing the error",
+    "message": "Unknown path components: /xyz",
+    "type": "OAuthException",
+    "code": 2500
+  }
+}`
+	tokenExpiredErrorBody = `
+{
+  "error": {
+    "message": "Error validating access token: Session has expired",
     "type": "OAuthException",
     "code": 190 ,
-    "error_subcode": 460
+    "error_subcode": 463
   }
 }`
 	notificationsBody = `
@@ -151,7 +161,17 @@
 	c.Check(notifications, IsNil)
 	c.Assert(err, Not(IsNil))
 	graphErr := err.(*GraphError)
-	c.Check(graphErr.Message, Equals, "Message describing the error")
-	c.Check(graphErr.Code, Equals, 190)
-	c.Check(graphErr.Subcode, Equals, 460)
+	c.Check(graphErr.Message, Equals, "Unknown path components: /xyz")
+	c.Check(graphErr.Code, Equals, 2500)
+}
+
+func (s S) TestTokenExpiredErrorResponse(c *C) {
+	resp := &http.Response{
+		StatusCode: http.StatusBadRequest,
+		Body:       closeWrapper{bytes.NewReader([]byte(tokenExpiredErrorBody))},
+	}
+	p := &fbPlugin{}
+	notifications, err := p.parseResponse(resp)
+	c.Check(notifications, IsNil)
+	c.Assert(err, Equals, plugins.ErrTokenExpired)
 }

=== modified file 'plugins/plugins.go'
--- plugins/plugins.go	2014-07-17 18:35:36 +0000
+++ plugins/plugins.go	2014-07-24 20:44:55 +0000
@@ -17,7 +17,11 @@
 
 package plugins
 
-import "launchpad.net/account-polld/accounts"
+import (
+	"errors"
+
+	"launchpad.net/account-polld/accounts"
+)
 
 // Plugin is an interface which the plugins will adhere to for the poll
 // daemon to interact with.
@@ -122,3 +126,7 @@
 	PLUGIN_EMAIL = 0
 	PLUGIN_SOCIAL
 )
+
+// ErrTokenExpired is the error returned by a plugin to indicate that
+// the web service reported that the authentication token has expired.
+var ErrTokenExpired = errors.New("Token expired")

=== modified file 'plugins/twitter/twitter.go'
--- plugins/twitter/twitter.go	2014-07-23 10:35:13 +0000
+++ plugins/twitter/twitter.go	2014-07-24 20:44:55 +0000
@@ -77,6 +77,12 @@
 		if err := decoder.Decode(&result); err != nil {
 			return nil, err
 		}
+		// Error code 89 is used for invalid or expired tokens.
+		for _, e := range result.Errors {
+			if e.Code == 89 {
+				return nil, plugins.ErrTokenExpired
+			}
+		}
 		return nil, &result
 	}
 
@@ -113,6 +119,12 @@
 		if err := decoder.Decode(&result); err != nil {
 			return nil, err
 		}
+		// Error code 89 is used for invalid or expired tokens.
+		for _, e := range result.Errors {
+			if e.Code == 89 {
+				return nil, plugins.ErrTokenExpired
+			}
+		}
 		return nil, &result
 	}
 

=== modified file 'plugins/twitter/twitter_test.go'
--- plugins/twitter/twitter_test.go	2014-07-18 11:37:04 +0000
+++ plugins/twitter/twitter_test.go	2014-07-24 20:44:55 +0000
@@ -23,6 +23,8 @@
 	"testing"
 
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/account-polld/plugins"
 )
 
 type S struct{}
@@ -54,6 +56,15 @@
     }
   ]
 }`
+	tokenExpiredErrorBody = `
+{
+  "errors": [
+    {
+      "message":"Invalid or expired token",
+      "code":89
+    }
+  ]
+}`
 	statusesBody = `
 [
   {
@@ -425,6 +436,17 @@
 	c.Check(twErr.Errors[0].Code, Equals, 34)
 }
 
+func (s S) TestParseStatusesTokenExpiredError(c *C) {
+	resp := &http.Response{
+		StatusCode: http.StatusBadRequest,
+		Body:       closeWrapper{bytes.NewReader([]byte(tokenExpiredErrorBody))},
+	}
+	p := &twitterPlugin{}
+	messages, err := p.parseStatuses(resp)
+	c.Check(messages, IsNil)
+	c.Assert(err, Equals, plugins.ErrTokenExpired)
+}
+
 func (s S) TestParseDirectMessages(c *C) {
 	resp := &http.Response{
 		StatusCode: http.StatusOK,
@@ -453,3 +475,14 @@
 	c.Check(twErr.Errors[0].Message, Equals, "Sorry, that page does not exist")
 	c.Check(twErr.Errors[0].Code, Equals, 34)
 }
+
+func (s S) TestParseDirectMessagesTokenExpiredError(c *C) {
+	resp := &http.Response{
+		StatusCode: http.StatusBadRequest,
+		Body:       closeWrapper{bytes.NewReader([]byte(tokenExpiredErrorBody))},
+	}
+	p := &twitterPlugin{}
+	messages, err := p.parseDirectMessages(resp)
+	c.Check(messages, IsNil)
+	c.Assert(err, Equals, plugins.ErrTokenExpired)
+}