vendor/whisper, statusd: push notifications implemented, closes #135

This commit is contained in:
Victor Farazdagi 2017-04-12 21:26:54 +03:00
parent ecea845d88
commit 92afd0d47e
15 changed files with 1479 additions and 26 deletions

View File

@ -31,9 +31,13 @@ var (
Name: "mailserver",
Usage: "Delivers expired messages on demand",
}
WhisperPassword = cli.StringFlag{
WhisperIdentityFile = cli.StringFlag{
Name: "identity",
Usage: "Protocol identity file (private key used for asymetric encryption)",
}
WhisperPasswordFile = cli.StringFlag{
Name: "password",
Usage: "Password, will be used for topic keys, as Mail & Notification Server password",
Usage: "Password file (password is used for symmetric encryption)",
}
WhisperPortFlag = cli.IntFlag{
Name: "port",
@ -54,6 +58,10 @@ var (
Name: "injectaccounts",
Usage: "Whether test account should be injected or not (default: true)",
}
FirebaseAuthorizationKey = cli.StringFlag{
Name: "firebaseauth",
Usage: "FCM Authorization Key used for sending Push Notifications",
}
)
var (
@ -67,11 +75,13 @@ var (
WhisperNotificationServerNodeFlag,
WhisperForwarderNodeFlag,
WhisperMailserverNodeFlag,
WhisperPassword,
WhisperIdentityFile,
WhisperPasswordFile,
WhisperPoWFlag,
WhisperPortFlag,
WhisperTTLFlag,
WhisperInjectTestAccounts,
FirebaseAuthorizationKey,
},
}
)
@ -129,24 +139,52 @@ func makeWhisperNodeConfig(ctx *cli.Context) (*params.NodeConfig, error) {
whisperConfig := nodeConfig.WhisperConfig
whisperConfig.Enabled = true
whisperConfig.IdentityFile = ctx.String(WhisperIdentityFile.Name)
whisperConfig.PasswordFile = ctx.String(WhisperPasswordFile.Name)
whisperConfig.EchoMode = ctx.BoolT(WhisperEchoModeFlag.Name)
whisperConfig.BootstrapNode = ctx.BoolT(WhisperBootstrapNodeFlag.Name)
whisperConfig.ForwarderNode = ctx.Bool(WhisperForwarderNodeFlag.Name)
whisperConfig.NotificationServerNode = ctx.Bool(WhisperNotificationServerNodeFlag.Name)
whisperConfig.MailServerNode = ctx.Bool(WhisperMailserverNodeFlag.Name)
whisperConfig.MailServerPassword = ctx.String(WhisperPassword.Name)
whisperConfig.NotificationServerPassword = ctx.String(WhisperPassword.Name) // the same for both mail and notification servers
whisperConfig.Port = ctx.Int(WhisperPortFlag.Name)
whisperConfig.TTL = ctx.Int(WhisperTTLFlag.Name)
whisperConfig.MinimumPoW = ctx.Float64(WhisperPoWFlag.Name)
if whisperConfig.MailServerNode && len(whisperConfig.MailServerPassword) == 0 {
if whisperConfig.MailServerNode && len(whisperConfig.PasswordFile) == 0 {
return nil, errors.New("mail server requires --password to be specified")
}
if whisperConfig.NotificationServerNode && len(whisperConfig.NotificationServerPassword) == 0 {
return nil, errors.New("notification server requires --password to be specified")
if whisperConfig.NotificationServerNode && len(whisperConfig.IdentityFile) == 0 {
return nil, errors.New("notification server requires either --identity file to be specified")
}
if len(whisperConfig.PasswordFile) > 0 { // make sure that we can load password file
if whisperConfig.PasswordFile, err = filepath.Abs(whisperConfig.PasswordFile); err != nil {
return nil, err
}
if _, err := whisperConfig.ReadPasswordFile(); err != nil {
return nil, err
}
}
if len(whisperConfig.IdentityFile) > 0 { // make sure that we can load identity file
if whisperConfig.IdentityFile, err = filepath.Abs(whisperConfig.IdentityFile); err != nil {
return nil, err
}
if _, err := whisperConfig.ReadIdentityFile(); err != nil {
return nil, err
}
}
firebaseConfig := whisperConfig.FirebaseConfig
firebaseConfig.AuthorizationKeyFile = ctx.String(FirebaseAuthorizationKey.Name)
if len(firebaseConfig.AuthorizationKeyFile) > 0 { // make sure authorization key can be loaded
if firebaseConfig.AuthorizationKeyFile, err = filepath.Abs(firebaseConfig.AuthorizationKeyFile); err != nil {
return nil, err
}
if _, err := firebaseConfig.ReadAuthorizationKeyFile(); err != nil {
return nil, err
}
}
return nodeConfig, nil

View File

@ -24,6 +24,8 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/discv5"
"github.com/ethereum/go-ethereum/p2p/nat"
"github.com/ethereum/go-ethereum/whisper/mailserver"
"github.com/ethereum/go-ethereum/whisper/notifications"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
"github.com/status-im/status-go/geth/params"
)
@ -114,6 +116,17 @@ func MakeNode(config *params.NodeConfig) *Node {
stackConfig.P2P.PrivateKey = pk
}
if len(config.NodeKeyFile) > 0 {
log.Info("Loading private key file", "file", config.NodeKeyFile)
pk, err := crypto.LoadECDSA(config.NodeKeyFile)
if err != nil {
log.Info("Failed loading private key file", "file", config.NodeKeyFile, "err", err)
}
// override node's private key
stackConfig.P2P.PrivateKey = pk
}
stack, err := node.New(stackConfig)
if err != nil {
Fatalf(ErrNodeMakeFailure)
@ -175,7 +188,30 @@ func activateShhService(stack *node.Node, config *params.NodeConfig) error {
return nil
}
serviceConstructor := func(*node.ServiceContext) (node.Service, error) {
return whisper.New(), nil
whisperConfig := config.WhisperConfig
whisperService := whisper.New()
// enable mail service
if whisperConfig.MailServerNode {
password, err := whisperConfig.ReadPasswordFile()
if err != nil {
return nil, err
}
var mailServer mailserver.WMailServer
whisperService.RegisterServer(&mailServer)
mailServer.Init(whisperService, whisperConfig.DataDir, string(password), whisperConfig.MinimumPoW)
}
// enable notification service
if whisperConfig.NotificationServerNode {
var notificationServer notifications.NotificationServer
whisperService.RegisterNotificationServer(&notificationServer)
notificationServer.Init(whisperService, whisperConfig)
}
return whisperService, nil
}
if err := stack.Register(serviceConstructor); err != nil {
return err
@ -286,7 +322,7 @@ func Fatalf(reason interface{}, args ...interface{}) {
// HaltOnPanic recovers from panic, logs issue, sends upward notification, and exits
func HaltOnPanic() {
if r := recover(); r != nil {
err := fmt.Errorf("%v: %v", ErrNodeStartFailure, r)
err := fmt.Errorf("%v: %v", ErrNodeRunFailure, r)
// send signal up to native app
SendSignal(SignalEnvelope{

View File

@ -53,6 +53,7 @@ var (
ErrInvalidJailedRequestQueue = errors.New("jailed request queue is not properly initialized")
ErrNodeMakeFailure = errors.New("error creating p2p node")
ErrNodeStartFailure = errors.New("error starting p2p node")
ErrNodeRunFailure = errors.New("error running p2p node")
ErrInvalidNodeAPI = errors.New("no node API connected")
ErrAccountKeyStoreMissing = errors.New("account key store is not set")
)

View File

@ -1,6 +1,8 @@
package params
import (
"bytes"
"crypto/ecdsa"
"encoding/json"
"errors"
"fmt"
@ -10,6 +12,7 @@ import (
"strings"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
)
@ -28,6 +31,9 @@ func init() {
var (
ErrMissingDataDir = errors.New("missing required 'DataDir' parameter")
ErrMissingNetworkId = errors.New("missing required 'NetworkId' parameter")
ErrEmptyPasswordFile = errors.New("password file cannot be empty")
ErrEmptyIdentityFile = errors.New("identity file cannot be empty")
ErrEmptyAuthorizationKeyFile = errors.New("authorization key file cannot be empty")
)
// LightEthConfig holds LES-related configuration
@ -43,11 +49,26 @@ type LightEthConfig struct {
DatabaseCache int
}
type FirebaseConfig struct {
// AuthorizationKeyFile file path that contains FCM authorization key
AuthorizationKeyFile string
// NotificationTriggerURL URL used to send push notification requests to
NotificationTriggerURL string
}
// WhisperConfig holds SHH-related configuration
type WhisperConfig struct {
// Enabled flag specifies whether protocol is enabled
Enabled bool
// IdentityFile path to private key, that will be loaded as identity into Whisper
IdentityFile string
// PasswordFile path to password file, for non-interactive password entry
// (if no account file selected, then this password is used for symmetric encryption)
PasswordFile string
// EchoMode if mode is on, prints some arguments for diagnostics
EchoMode bool
@ -60,15 +81,9 @@ type WhisperConfig struct {
// MailServerNode is mode when node is capable of delivering expired messages on demand
MailServerNode bool
// MailServerPassword is password for MailServer's symmetric key
MailServerPassword string
// NotificationServerNode is mode when node is capable of sending Push (and probably other kinds) Notifications
NotificationServerNode bool
// NotificationServerPassword is password for NotificationServer's symmetric key (used in discovery)
NotificationServerPassword string
// DataDir is the file system folder Whisper should use for any data storage needs.
DataDir string
@ -80,6 +95,9 @@ type WhisperConfig struct {
// TTL time to live for messages, in seconds
TTL int
// FirebaseConfig extra configuration for Firebase Cloud Messaging
FirebaseConfig *FirebaseConfig `json:"FirebaseConfig,"`
}
// SwarmConfig holds Swarm-related configuration
@ -200,6 +218,9 @@ func NewNodeConfig(dataDir string, networkId uint64) (*NodeConfig, error) {
Port: WhisperPort,
MinimumPoW: WhisperMinimumPoW,
TTL: WhisperTTL,
FirebaseConfig: &FirebaseConfig{
NotificationTriggerURL: FirebaseNotificationTriggerURL,
},
},
SwarmConfig: &SwarmConfig{},
}
@ -330,3 +351,59 @@ func (c *SwarmConfig) String() string {
data, _ := json.MarshalIndent(c, "", " ")
return string(data)
}
// ReadPasswordFile reads and returns content of the password file
func (c *WhisperConfig) ReadPasswordFile() ([]byte, error) {
if len(c.PasswordFile) <= 0 {
return nil, ErrEmptyPasswordFile
}
password, err := ioutil.ReadFile(c.PasswordFile)
if err != nil {
return nil, err
}
password = bytes.TrimRight(password, "\n")
if len(password) == 0 {
return nil, ErrEmptyPasswordFile
}
return password, nil
}
// ReadIdentityFile reads and loads identity private key
func (c *WhisperConfig) ReadIdentityFile() (*ecdsa.PrivateKey, error) {
if len(c.IdentityFile) <= 0 {
return nil, ErrEmptyIdentityFile
}
identity, err := crypto.LoadECDSA(c.IdentityFile)
if err != nil {
return nil, err
}
if identity == nil {
return nil, ErrEmptyIdentityFile
}
return identity, nil
}
// ReadAuthorizationKeyFile reads and loads FCM authorization key
func (c *FirebaseConfig) ReadAuthorizationKeyFile() ([]byte, error) {
if len(c.AuthorizationKeyFile) <= 0 {
return nil, ErrEmptyAuthorizationKeyFile
}
key, err := ioutil.ReadFile(c.AuthorizationKeyFile)
if err != nil {
return nil, err
}
key = bytes.TrimRight(key, "\n")
if key == nil {
return nil, ErrEmptyAuthorizationKeyFile
}
return key, nil
}

View File

@ -66,6 +66,9 @@ const (
// WhisperTTL is time to live for messages, in seconds
WhisperTTL = 120
// FirebaseNotificationTriggerURL is URL where FCM notification requests are sent to
FirebaseNotificationTriggerURL = "https://fcm.googleapis.com/fcm/send"
// MainNetworkId is id of the main network
MainNetworkId = 1

View File

@ -28,17 +28,21 @@
},
"WhisperConfig": {
"Enabled": true,
"IdentityFile": "",
"PasswordFile": "",
"EchoMode": false,
"BootstrapNode": false,
"ForwarderNode": false,
"MailServerNode": false,
"MailServerPassword": "",
"NotificationServerNode": false,
"NotificationServerPassword": "",
"DataDir": "$TMPDIR/wnode",
"Port": 30379,
"MinimumPoW": 0.001,
"TTL": 120
"TTL": 120,
"FirebaseConfig": {
"AuthorizationKeyFile": "",
"NotificationTriggerURL": "https://fcm.googleapis.com/fcm/send"
}
},
"SwarmConfig": {
"Enabled": false

View File

@ -28,17 +28,21 @@
},
"WhisperConfig": {
"Enabled": true,
"IdentityFile": "",
"PasswordFile": "",
"EchoMode": false,
"BootstrapNode": false,
"ForwarderNode": false,
"MailServerNode": false,
"MailServerPassword": "",
"NotificationServerNode": false,
"NotificationServerPassword": "",
"DataDir": "$TMPDIR/wnode",
"Port": 30379,
"MinimumPoW": 0.001,
"TTL": 120
"TTL": 120,
"FirebaseConfig": {
"AuthorizationKeyFile": "",
"NotificationTriggerURL": "https://fcm.googleapis.com/fcm/send"
}
},
"SwarmConfig": {
"Enabled": false

View File

@ -0,0 +1,392 @@
<html lang="en">
<head>
<title>Whisper Notification Server Test</title>
<link rel="stylesheet" href="http://maxcdn.bootstrapcdn.com/bootstrap/3.2.0/css/bootstrap.min.css">
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/font-awesome/4.7.0/css/font-awesome.min.css">
<script
src="https://code.jquery.com/jquery-3.1.1.js"
integrity="sha256-16cdPddA6VdVInumRGo6IbivbERE8p7CQR3HzTBuELA="
crossorigin="anonymous"></script>
<script type="text/javascript" src="http://maxcdn.bootstrapcdn.com/bootstrap/3.2.0/js/bootstrap.min.js"></script>
<script type="text/javascript" src="./../scripts/bignumber.js"></script>
<script type="text/javascript" src="./../scripts/web3.js"></script>
<script type="text/javascript">
var protocolKey = '0x040edb0d71a3dbe928e154fcb696ffbda359b153a90efc2b46f0043ce9f5dbe55b77b9328fd841a1db5273758624afadd5b39638d4c35b36b3a96e1a586c1b4c2a';
var discoverServerTopic = '0x268302f3'; // DISCOVER_NOTIFICATION_SERVER
var proposeServerTopic = '0x08e3d8c0'; // PROPOSE_NOTIFICATION_SERVER
var acceptServerTopic = '0x04f7dea6'; // ACCEPT_NOTIFICATION_SERVER
var ackClientSubscriptionTopic = '0x93dafe28'; // ACK_NOTIFICATION_SERVER_SUBSCRIPTION
var sendNotificationTopic = '0x69915296'; // SEND_NOTIFICATION
var newChatSessionTopic = '0x509579a2'; // NEW_CHAT_SESSION
var ackNewChatSessionTopic = '0xd012aae8'; // ACK_NEW_CHAT_SESSION
var newDeviceRegistrationTopic = '0x14621a51'; // NEW_DEVICE_REGISTRATION
var ackDeviceRegistrationTopic = '0x424358d6'; // ACK_DEVICE_REGISTRATION
var checkClientSessionTopic = '0x8745d931'; // CHECK_CLIENT_SESSION
var confirmClientSessionTopic = '0xd3202c5f'; // CONFIRM_CLIENT_SESSION
var dropClientSessionTopic = '0x3a6656bb'; // DROP_CLIENT_SESSION
// this will be used both by Device A and Device B
var registerDevice = function (web3, identity, chatId, chatKey, deviceId) {
console.log('chat session key: ', chatKey);
// make sure that chat key is loaded
var keyname = chatId + 'chatkey'; // there might be many chat keys
web3.shh.deleteSymKey(keyname);
web3.shh.addSymKey(keyname, chatKey);
// before sending request, let's start waiting for response
var filter = web3.shh.filter({
to: identity,
topics: [ackDeviceRegistrationTopic]
});
filter.watch(function (error, result) {
if (!error) {
// response will be in JSON, e.g. {"server": "0xdeadbeef"}
var payload = JSON.parse(web3.toAscii(result.payload));
console.log("Device Registration ACK received: ", result, payload);
// no need to watch for the filter any more
filter.stopWatching();
}
});
var err = web3.shh.post({
from: identity,
topics: [newDeviceRegistrationTopic],
payload: '{"device": "' + deviceId + '"}',
ttl: 20,
keyname: keyname
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
};
var startDeviceA = function () {
var web3 = new Web3();
web3.setProvider(new web3.providers.HttpProvider('http://localhost:8645'));
var sendDiscoveryRequest = function (identity) {
// notification server discovery request is a signed (sent from us),
// encrypted with Notification Protocol Asymmetric (Public) Key
var err = web3.shh.post({
from: identity,
to: protocolKey,
topics: [discoverServerTopic],
ttl: 20
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
};
var sendAcceptServerRequest = function (identity, serverId) {
// whenever we are ready to accept server, having a given serverId, we need
// to notify it by sending signed (from us) and encrypted (using protocol key)
// acceptance message.
var err = web3.shh.post({
from: identity, // it is absolutely important to identify the client, or your acceptance will be dropped
to: protocolKey,
topics: [acceptServerTopic],
payload: '{"server": "' + serverId + '"}',
ttl: 20
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
};
var watchProposeServerResponses = function (identity) {
// some notification servers will be able to serve, so they will send encrypted (to you)
// message, with a PROPOSE_NOTIFICATION_SERVER topic (for which we wait)
var filter = web3.shh.filter({
to: identity, // wait for anon. messages to ourselves
topics: [proposeServerTopic]
});
filter.watch(function (error, result) {
if (!error) {
console.log("Server proposal received: ", result);
// response will be in JSON, e.g. {"server": "0x81f34abd0df038e01a8f9c04bee7ce92925b7240e334dc8f2400dea7a2a6d829678be8b40e1d9b9988e25960552eafe2df7f928188e4143ba657a699519c438d"}
// which will give you serverId
var payload = JSON.parse(web3.toAscii(result.payload));
console.log(payload);
// no need to watch for the filter any more
filter.stopWatching();
// accept (in FIFO order) the server
// we need to make sure that only a single server is selected,
// as you will receive multiple proposals for different servers,
// and may accept more that one of those proposals (which will
// result in duplicate notifications)
sendAcceptServerRequest(identity, payload.server);
}
});
};
var shareChatKey = function (chatId, chatKey) {
console.log('chat session key: ', chatKey)
// pre-defined test identity (it gets injected automatically by statusd)
var deviceBIdentity = '0x04eedbaafd6adf4a9233a13e7b1c3c14461fffeba2e9054b8d456ce5f6ebeafadcbf3dce3716253fbc391277fa5a086b60b283daf61fb5b1f26895f456c2f31ae3';
// it is up to you how you share secret among participants, here is sample
var err = web3.shh.post({
from: identity,
to: deviceBIdentity,
topics: ["chatKeySharing"],
payload: '{"chat": "' + chatId + '", "key": "' + chatKey + '"}',
ttl: 20
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
};
var removeClientSession = function () {
var checkClientSession = function (callback) {
// before sending request, let's start waiting for response
var filter = web3.shh.filter({
to: identity,
topics: [confirmClientSessionTopic]
});
filter.watch(function (error, result) {
if (!error) {
// response will be in JSON, e.g. {"server": "0xdeadbeef", "key": "0xdeadbeef"}
var payload = JSON.parse(web3.toAscii(result.payload));
console.log("Client session confirmation received: ", result, payload);
// no need to watch for the filter any more
filter.stopWatching();
callback(payload)
}
});
// send enquiry to all servers, to learn whether we have a client session with any of them
var err = web3.shh.post({
from: identity,
to: protocolKey,
topics: [checkClientSessionTopic],
ttl: 20
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
};
checkClientSession(function (payload) {
console.log("time to cleanup: ", payload.server);
console.log("session key: ", payload.key);
setTimeout(function () {
// notify server that you want to unsubscribe
var err = web3.shh.post({
from: identity,
to: protocolKey,
topics: [dropClientSessionTopic],
ttl: 20
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
}, 5000); // let's all other concurrent tasks to wrap up
});
};
var createChatSession = function (subscriptionKey) {
var chatId = '0xdeadbeef';
// subscriptionKey is key shared by server that allows us to communicate with server privately
var keyname = 'SUBSCRIPTION_KEY'; // you might want to be tad more creative
web3.shh.deleteSymKey(keyname);
web3.shh.addSymKey(keyname, subscriptionKey);
console.log("subscription key: ", subscriptionKey);
// before sending new chat request, let's start waiting for response
var filter = web3.shh.filter({
to: identity,
topics: [ackNewChatSessionTopic]
});
filter.watch(function (error, result) {
if (!error) {
console.log("Chat Creation ACK received: ", result);
// response will be in JSON, e.g. {"server": "0xdeadbeef", "key": "0xdeadbeef"}
// which will give you serverId
var payload = JSON.parse(web3.toAscii(result.payload));
// no need to watch for the filter any more
filter.stopWatching();
// ok, at this point we have Chat Session SymKey, and we can:
// 1. register our device with that chat
// 2. share that key with others, so that they can register themselves
// 3. use chat key to trigger notifications
// this obtained from https://status-sandbox-c1b34.firebaseapp.com/
var deviceId = 'ca5pRJc6L8s:APA91bHpYFtpxvXx6uOayGmnNVnktA4PEEZdquCCt3fWR5ldLzSy1A37Tsbzk5Gavlmk1d_fvHRVnK7xPAhFFl-erF7O87DnIEstW6DEyhyiKZYA4dXFh6uy323f9A3uw5hEtT_kQVhT';
registerDevice(web3, identity, chatId, payload.key, deviceId); // weird signature because we reuse method for Device B
shareChatKey(chatId, payload.key);
// now do a cleanup
removeClientSession();
}
});
var err = web3.shh.post({
from: identity,
topics: [newChatSessionTopic],
payload: '{"chat": "' + chatId + '"}', // globally unique chat Id
ttl: 20,
keyname: keyname
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
};
var watchServerAckResponses = function (identity) {
// if server we accepted is ok, it will send encrypted (to you)
// message, with a ACK_NOTIFICATION_SERVER_SUBSCRIPTION topic (for which we wait)
// This message completes the subscription process. At this point you should
// have topic and symkey necessary to manage your subscription.
var filter = web3.shh.filter({
to: identity, // wait for anon. messages to ourselves
topics: [ackClientSubscriptionTopic]
});
filter.watch(function (error, result) {
if (!error) {
console.log("Server ACK received: ", result);
// response will be in JSON, e.g. {"server": "0xdeadbeef", "key": "0xdeadbeef"}
// which will give you serverId
var payload = JSON.parse(web3.toAscii(result.payload));
console.log(payload);
// no need to watch for the filter any more
filter.stopWatching();
// this concludes discovery, and we can use obtained key to invoke chat sessions
createChatSession(payload.key)
}
});
};
var identity = web3.shh.newIdentity();
console.log("identity used: ", identity)
// check identity
if (!web3.shh.hasIdentity(identity)) {
throw 'idenitity "' + identity + '" not found in whisper';
}
watchProposeServerResponses(identity);
watchServerAckResponses(identity);
// start discovery protocol, by sending discovery request
sendDiscoveryRequest(identity);
};
var startDeviceB = function () {
var web3 = new Web3();
web3.setProvider(new web3.providers.HttpProvider('http://localhost:8745'));
// pre-defined test identity (it gets injected automatically by statusd)
var identity = '0x04eedbaafd6adf4a9233a13e7b1c3c14461fffeba2e9054b8d456ce5f6ebeafadcbf3dce3716253fbc391277fa5a086b60b283daf61fb5b1f26895f456c2f31ae3';
if (!web3.shh.hasIdentity(identity)) {
throw 'idenitity "' + identity + '" not found in whisper';
}
// for for key sharing, it is up to you how you implement it (which topic to use etc)
var filter = web3.shh.filter({
to: identity, // wait for anon. messages to ourselves
topics: ['chatKeySharing']
});
filter.watch(function (error, result) {
if (!error) {
console.log("Chat key received: ", result);
// response will be in JSON, e.g. {chat: "0xdeadbeef", key: "0x04e68e37433baf55ddc2fe9f7533e4e722bcdad4239c98df92f3522907ced72d"}
var payload = JSON.parse(web3.toAscii(result.payload));
console.log(payload);
// no need to watch for the filter any more
filter.stopWatching();
// let's save incoming key
var keyname = payload.chat + '-chatkey';
web3.shh.deleteSymKey(keyname);
web3.shh.addSymKey(keyname, payload.key);
// now register ourselves
var deviceId = 'some-device-id'; // you obtain this from FCM
registerDevice(web3, identity, payload.chat, payload.key, deviceId);
// finally, trigger the notifications on all registered device (except for yourself)
// at this point it is really trivial, use the key + specific topic:
var err = web3.shh.post({
from: identity,
topics: [sendNotificationTopic],
payload: '{' // see https://firebase.google.com/docs/cloud-messaging/http-server-ref
+ '"notification": {'
+ '"title": "status.im notification",'
+ '"body": "Hello this is test notification!",'
+ '"icon": "https://status.im/img/logo.png",'
+ '"click_action": "https://status.im"'
+ '},'
+ '"to": "{{ ID }}"' // this get replaced by device id your've registered
+ '}',
ttl: 20,
keyname: keyname
});
if (err !== null) {
console.log("message NOT sent")
} else {
console.log("message sent OK")
}
}
});
console.log("device b started")
};
startDeviceA();
startDeviceB();
$(document).ready(function () {
});
</script>
</head>
<body>
<!-- Static navbar -->
<nav class="navbar navbar-default navbar-static-top">
<div class="container">
</div>
</nav>
<div class="container">
<div class="jumbotron">
</div>
</div>
</body>
</html>

View File

@ -0,0 +1,148 @@
package notifications
import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
)
const (
topicDiscoverServer = "DISCOVER_NOTIFICATION_SERVER"
topicProposeServer = "PROPOSE_NOTIFICATION_SERVER"
topicServerAccepted = "ACCEPT_NOTIFICATION_SERVER"
topicAckClientSubscription = "ACK_NOTIFICATION_SERVER_SUBSCRIPTION"
)
// discoveryService abstract notification server discovery protocol
type discoveryService struct {
server *NotificationServer
discoverFilterID string
serverAcceptedFilterID string
}
// messageProcessingFn is a callback used to process incoming client requests
type messageProcessingFn func(*whisper.ReceivedMessage) error
func NewDiscoveryService(notificationServer *NotificationServer) *discoveryService {
return &discoveryService{
server: notificationServer,
}
}
// Start installs necessary filters to watch for incoming discovery requests,
// then in separate routine starts watcher loop
func (s *discoveryService) Start() error {
var err error
// notification server discovery requests
s.discoverFilterID, err = s.server.installKeyFilter(topicDiscoverServer, s.server.protocolKey)
if err != nil {
return fmt.Errorf("failed installing filter: %v", err)
}
go s.server.requestProcessorLoop(s.discoverFilterID, topicDiscoverServer, s.processDiscoveryRequest)
// notification server accept/select requests
s.serverAcceptedFilterID, err = s.server.installKeyFilter(topicServerAccepted, s.server.protocolKey)
if err != nil {
return fmt.Errorf("failed installing filter: %v", err)
}
go s.server.requestProcessorLoop(s.serverAcceptedFilterID, topicServerAccepted, s.processServerAcceptedRequest)
log.Info("notification server discovery service started")
return nil
}
// Stop stops all discovery processing loops
func (s *discoveryService) Stop() error {
s.server.whisper.Unsubscribe(s.discoverFilterID)
s.server.whisper.Unsubscribe(s.serverAcceptedFilterID)
log.Info("notification server discovery service stopped")
return nil
}
// processDiscoveryRequest processes incoming client requests of type:
// when client tries to discover suitable notification server
func (s *discoveryService) processDiscoveryRequest(msg *whisper.ReceivedMessage) error {
// offer this node as notification server
msgParams := whisper.MessageParams{
Src: s.server.protocolKey,
Dst: msg.Src,
Topic: MakeTopic([]byte(topicProposeServer)),
Payload: []byte(`{"server": "0x` + s.server.nodeID + `"}`),
TTL: uint32(s.server.config.TTL),
PoW: s.server.config.MinimumPoW,
WorkTime: 5,
}
response := whisper.NewSentMessage(&msgParams)
env, err := response.Wrap(&msgParams)
if err != nil {
return fmt.Errorf("failed to wrap server proposal message: %v", err)
}
if err := s.server.whisper.Send(env); err != nil {
return fmt.Errorf("failed to send server proposal message: %v", err)
}
log.Info(fmt.Sprintf("server proposal sent (server: %v, dst: %v, topic: %x)",
s.server.nodeID, common.ToHex(crypto.FromECDSAPub(msgParams.Dst)), msgParams.Topic))
return nil
}
// processServerAcceptedRequest processes incoming client requests of type:
// when client is ready to select the given node as its notification server
func (s *discoveryService) processServerAcceptedRequest(msg *whisper.ReceivedMessage) error {
var parsedMessage struct {
ServerID string `json:"server"`
}
if err := json.Unmarshal(msg.Payload, &parsedMessage); err != nil {
return err
}
if msg.Src == nil {
return errors.New("message 'from' field is required")
}
// make sure that only requests made to the current node are processed
if parsedMessage.ServerID != `0x`+s.server.nodeID {
return nil
}
// register client
sessionKey, err := s.server.RegisterClientSession(&ClientSession{
ClientKey: hex.EncodeToString(crypto.FromECDSAPub(msg.Src)),
})
if err != nil {
return err
}
// confirm that client has been successfully subscribed
msgParams := whisper.MessageParams{
Src: s.server.protocolKey,
Dst: msg.Src,
Topic: MakeTopic([]byte(topicAckClientSubscription)),
Payload: []byte(`{"server": "0x` + s.server.nodeID + `", "key": "0x` + hex.EncodeToString(sessionKey) + `"}`),
TTL: uint32(s.server.config.TTL),
PoW: s.server.config.MinimumPoW,
WorkTime: 5,
}
response := whisper.NewSentMessage(&msgParams)
env, err := response.Wrap(&msgParams)
if err != nil {
return fmt.Errorf("failed to wrap server proposal message: %v", err)
}
if err := s.server.whisper.Send(env); err != nil {
return fmt.Errorf("failed to send server proposal message: %v", err)
}
log.Info(fmt.Sprintf("server confirms client subscription (dst: %v, topic: %x)", msgParams.Dst, msgParams.Topic))
return nil
}

View File

@ -0,0 +1,59 @@
package notifications
import (
"bytes"
"fmt"
"io/ioutil"
"net/http"
"strings"
"github.com/ethereum/go-ethereum/log"
"github.com/status-im/status-go/geth/params"
)
// NotificationDeliveryProvider handles the notification delivery
type NotificationDeliveryProvider interface {
Send(id string, payload string) error
}
// FirebaseProvider represents FCM provider
type FirebaseProvider struct {
AuthorizationKey string
NotificationTriggerURL string
}
// NewFirebaseProvider creates new FCM provider
func NewFirebaseProvider(config *params.FirebaseConfig) *FirebaseProvider {
authorizationKey, _ := config.ReadAuthorizationKeyFile()
return &FirebaseProvider{
NotificationTriggerURL: config.NotificationTriggerURL,
AuthorizationKey: string(authorizationKey),
}
}
// Send triggers sending of Push Notification to a given device id
func (p *FirebaseProvider) Send(id string, payload string) (err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("panic: %v", r)
}
}()
jsonRequest := strings.Replace(payload, "{{ ID }}", id, 3)
req, err := http.NewRequest("POST", p.NotificationTriggerURL, bytes.NewBuffer([]byte(jsonRequest)))
req.Header.Set("Authorization", "key="+p.AuthorizationKey)
req.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
log.Debug("FCM response", "status", resp.Status, "header", resp.Header)
body, _ := ioutil.ReadAll(resp.Body)
log.Debug("FCM response body", "body", string(body))
return nil
}

View File

@ -0,0 +1,581 @@
package notifications
import (
"errors"
"fmt"
"sync"
"time"
"crypto/ecdsa"
"encoding/hex"
"encoding/json"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
"github.com/status-im/status-go/geth/params"
)
const (
topicSendNotification = "SEND_NOTIFICATION"
topicNewChatSession = "NEW_CHAT_SESSION"
topicAckNewChatSession = "ACK_NEW_CHAT_SESSION"
topicNewDeviceRegistration = "NEW_DEVICE_REGISTRATION"
topicAckDeviceRegistration = "ACK_DEVICE_REGISTRATION"
topicCheckClientSession = "CHECK_CLIENT_SESSION"
topicConfirmClientSession = "CONFIRM_CLIENT_SESSION"
topicDropClientSession = "DROP_CLIENT_SESSION"
)
var (
ErrServiceInitError = errors.New("notification service has not been properly initialized")
)
// NotificationServer service capable of handling Push Notifications
type NotificationServer struct {
whisper *whisper.Whisper
config *params.WhisperConfig
nodeID string // proposed server will feature this ID
discovery *discoveryService // discovery service handles client/server negotiation, when server is selected
protocolKey *ecdsa.PrivateKey // private key of service, used to encode handshake communication
clientSessions map[string]*ClientSession
clientSessionsMu sync.RWMutex
chatSessions map[string]*ChatSession
chatSessionsMu sync.RWMutex
deviceSubscriptions map[string]*DeviceSubscription
deviceSubscriptionsMu sync.RWMutex
firebaseProvider NotificationDeliveryProvider
quit chan struct{}
}
// ClientSession abstracts notification client, which expects notifications whenever
// some envelope can be decoded with session key (key hash is compared for optimization)
type ClientSession struct {
ClientKey string // public key uniquely identifying a client
SessionKey []byte // actual symkey used for client - server communication
SessionKeyHash common.Hash // The Keccak256Hash of the symmetric key, which is shared between server/client
SessionKeyInput []byte // raw symkey used as input for actual SessionKey
}
// ChatSession abstracts chat session, which some previously registered client can create.
// ChatSession is used by client for sharing common secret, allowing others to register
// themselves and eventually to trigger notifications.
type ChatSession struct {
ParentKey string // public key uniquely identifying a client session used to create a chat session
ChatKey string // ID that uniquely identifies a chat session
SessionKey []byte // actual symkey used for client - server communication
SessionKeyHash common.Hash // The Keccak256Hash of the symmetric key, which is shared between server/client
}
// DeviceSubscription stores enough information about a device (or group of devices),
// so that Notification Server can trigger notification on that device(s)
type DeviceSubscription struct {
DeviceID string // ID that will be used as destination
ChatSessionKeyHash common.Hash // The Keccak256Hash of the symmetric key, which is shared between server/client
PubKey *ecdsa.PublicKey // public key of subscriber (to filter out when notification is triggered)
}
// Init used for service initialization, making sure it is safe to call Start()
func (s *NotificationServer) Init(whisperService *whisper.Whisper, whisperConfig *params.WhisperConfig) {
s.whisper = whisperService
s.config = whisperConfig
s.discovery = NewDiscoveryService(s)
s.clientSessions = make(map[string]*ClientSession)
s.chatSessions = make(map[string]*ChatSession)
s.deviceSubscriptions = make(map[string]*DeviceSubscription)
s.quit = make(chan struct{})
// setup providers (FCM only, for now)
s.firebaseProvider = NewFirebaseProvider(whisperConfig.FirebaseConfig)
}
// Start begins notification loop, in a separate go routine
func (s *NotificationServer) Start(stack *p2p.Server) error {
if s.whisper == nil {
return ErrServiceInitError
}
// configure nodeID
if stack != nil {
if nodeInfo := stack.NodeInfo(); nodeInfo != nil {
s.nodeID = nodeInfo.ID
}
}
// configure keys
identity, err := s.config.ReadIdentityFile()
if err != nil {
return err
}
s.whisper.AddKeyPair(identity)
s.protocolKey = identity
log.Info("protocol pubkey", "key", common.ToHex(crypto.FromECDSAPub(&s.protocolKey.PublicKey)))
// start discovery protocol
s.discovery.Start()
// client session status requests
clientSessionStatusFilterID, err := s.installKeyFilter(topicCheckClientSession, s.protocolKey)
if err != nil {
return fmt.Errorf("failed installing filter: %v", err)
}
go s.requestProcessorLoop(clientSessionStatusFilterID, topicDiscoverServer, s.processClientSessionStatusRequest)
// client session remove requests
dropClientSessionFilterID, err := s.installKeyFilter(topicDropClientSession, s.protocolKey)
if err != nil {
return fmt.Errorf("failed installing filter: %v", err)
}
go s.requestProcessorLoop(dropClientSessionFilterID, topicDropClientSession, s.processDropClientSessionRequest)
log.Info("Whisper Notification Server started")
return nil
}
// Stop handles stopping the running notification loop, and all related resources
func (s *NotificationServer) Stop() error {
close(s.quit)
if s.whisper == nil {
return ErrServiceInitError
}
if s.discovery != nil {
s.discovery.Stop()
}
log.Info("Whisper Notification Server stopped")
return nil
}
// RegisterClientSession forms a cryptographic link between server and client.
// It does so by sharing a session SymKey and installing filter listening for messages
// encrypted with that key. So, both server and client have a secure way to communicate.
func (s *NotificationServer) RegisterClientSession(session *ClientSession) (sessionKey []byte, err error) {
s.clientSessionsMu.Lock()
defer s.clientSessionsMu.Unlock()
// generate random symmetric session key
keyName := fmt.Sprintf("%s-%s", "ntfy-client", crypto.Keccak256Hash([]byte(session.ClientKey)).Hex())
sessionKey, sessionKeyDerived, err := s.makeSessionKey(keyName)
if err != nil {
return nil, err
}
// populate session key hash (will be used to match decrypted message to a given client id)
session.SessionKeyInput = sessionKey
session.SessionKeyHash = crypto.Keccak256Hash(sessionKeyDerived)
session.SessionKey = sessionKeyDerived
// append to list of known clients
// so that it is trivial to go key hash -> client session info
id := session.SessionKeyHash.Hex()
s.clientSessions[id] = session
// setup filter, which will get all incoming messages, that are encrypted with SymKey
filterID, err := s.installTopicFilter(topicNewChatSession, sessionKeyDerived)
if err != nil {
return nil, fmt.Errorf("failed installing filter: %v", err)
}
go s.requestProcessorLoop(filterID, topicNewChatSession, s.processNewChatSessionRequest)
return
}
// RegisterChatSession forms a cryptographic link between server and client.
// This link is meant to be shared with other clients, so that they can use
// the shared SymKey to trigger notifications for devices attached to a given
// chat session.
func (s *NotificationServer) RegisterChatSession(session *ChatSession) (sessionKey []byte, err error) {
s.chatSessionsMu.Lock()
defer s.chatSessionsMu.Unlock()
// generate random symmetric session key
keyName := fmt.Sprintf("%s-%s", "ntfy-chat", crypto.Keccak256Hash([]byte(session.ParentKey+session.ChatKey)).Hex())
sessionKey, sessionKeyDerived, err := s.makeSessionKey(keyName)
if err != nil {
return nil, err
}
// populate session key hash (will be used to match decrypted message to a given client id)
session.SessionKeyHash = crypto.Keccak256Hash(sessionKeyDerived)
session.SessionKey = sessionKeyDerived
// append to list of known clients
// so that it is trivial to go key hash -> client session info
id := session.SessionKeyHash.Hex()
s.chatSessions[id] = session
// setup filter, to process incoming device registration requests
filterID1, err := s.installTopicFilter(topicNewDeviceRegistration, sessionKeyDerived)
if err != nil {
return nil, fmt.Errorf("failed installing filter: %v", err)
}
go s.requestProcessorLoop(filterID1, topicNewDeviceRegistration, s.processNewDeviceRegistrationRequest)
// setup filter, to process incoming notification trigger requests
filterID2, err := s.installTopicFilter(topicSendNotification, sessionKeyDerived)
if err != nil {
return nil, fmt.Errorf("failed installing filter: %v", err)
}
go s.requestProcessorLoop(filterID2, topicSendNotification, s.processSendNotificationRequest)
return
}
// RegisterDeviceSubscription persists device id, so that it can be used to trigger notifications.
func (s *NotificationServer) RegisterDeviceSubscription(subscription *DeviceSubscription) error {
s.deviceSubscriptionsMu.Lock()
defer s.deviceSubscriptionsMu.Unlock()
// if one passes the same id again, we will just overwrite
id := fmt.Sprintf("%s-%s", "ntfy-device",
crypto.Keccak256Hash([]byte(subscription.ChatSessionKeyHash.Hex()+subscription.DeviceID)).Hex())
s.deviceSubscriptions[id] = subscription
log.Info("device registered", "device", subscription.DeviceID)
return nil
}
// DropClientSession uninstalls session
func (s *NotificationServer) DropClientSession(id string) {
dropChatSessions := func(parentKey string) {
s.chatSessionsMu.Lock()
defer s.chatSessionsMu.Unlock()
for key, chatSession := range s.chatSessions {
if chatSession.ParentKey == parentKey {
delete(s.chatSessions, key)
log.Info("drop chat session", "key", key)
}
}
}
dropDeviceSubscriptions := func(parentKey string) {
s.deviceSubscriptionsMu.Lock()
defer s.deviceSubscriptionsMu.Unlock()
for key, subscription := range s.deviceSubscriptions {
if hex.EncodeToString(crypto.FromECDSAPub(subscription.PubKey)) == parentKey {
delete(s.deviceSubscriptions, key)
log.Info("drop device subscription", "key", key)
}
}
}
s.clientSessionsMu.Lock()
if session, ok := s.clientSessions[id]; ok {
delete(s.clientSessions, id)
log.Info("server drops client session", "id", id)
s.clientSessionsMu.Unlock()
dropDeviceSubscriptions(session.ClientKey)
dropChatSessions(session.ClientKey)
}
}
// processNewChatSessionRequest processes incoming client requests of type:
// client has a session key, and ready to create a new chat session (which is
// a bag of subscribed devices, basically)
func (s *NotificationServer) processNewChatSessionRequest(msg *whisper.ReceivedMessage) error {
s.clientSessionsMu.RLock()
defer s.clientSessionsMu.RUnlock()
var parsedMessage struct {
ChatID string `json:"chat"`
}
if err := json.Unmarshal(msg.Payload, &parsedMessage); err != nil {
return err
}
if msg.Src == nil {
return errors.New("message 'from' field is required")
}
clientSession, ok := s.clientSessions[msg.SymKeyHash.Hex()]
if !ok {
return errors.New("client session not found")
}
// register chat session
parentKey := hex.EncodeToString(crypto.FromECDSAPub(msg.Src))
sessionKey, err := s.RegisterChatSession(&ChatSession{
ParentKey: parentKey,
ChatKey: parsedMessage.ChatID,
})
if err != nil {
return err
}
// confirm that chat has been successfully created
msgParams := whisper.MessageParams{
Dst: msg.Src,
KeySym: clientSession.SessionKey,
Topic: MakeTopic([]byte(topicAckNewChatSession)),
Payload: []byte(`{"server": "0x` + s.nodeID + `", "key": "0x` + hex.EncodeToString(sessionKey) + `"}`),
TTL: uint32(s.config.TTL),
PoW: s.config.MinimumPoW,
WorkTime: 5,
}
response := whisper.NewSentMessage(&msgParams)
env, err := response.Wrap(&msgParams)
if err != nil {
return fmt.Errorf("failed to wrap server response message: %v", err)
}
if err := s.whisper.Send(env); err != nil {
return fmt.Errorf("failed to send server response message: %v", err)
}
log.Info("server confirms chat creation", "dst",
common.ToHex(crypto.FromECDSAPub(msgParams.Dst)), "topic", msgParams.Topic.String())
return nil
}
// processNewDeviceRegistrationRequest processes incoming client requests of type:
// client has a session key, creates chat, and obtains chat SymKey (to be shared with
// others). Then using that chat SymKey client registers it's device ID with server.
func (s *NotificationServer) processNewDeviceRegistrationRequest(msg *whisper.ReceivedMessage) error {
s.chatSessionsMu.RLock()
defer s.chatSessionsMu.RUnlock()
var parsedMessage struct {
DeviceID string `json:"device"`
}
if err := json.Unmarshal(msg.Payload, &parsedMessage); err != nil {
return err
}
if msg.Src == nil {
return errors.New("message 'from' field is required")
}
chatSession, ok := s.chatSessions[msg.SymKeyHash.Hex()]
if !ok {
return errors.New("chat session not found")
}
if len(parsedMessage.DeviceID) <= 0 {
return errors.New("'device' cannot be empty")
}
// register chat session
err := s.RegisterDeviceSubscription(&DeviceSubscription{
DeviceID: parsedMessage.DeviceID,
ChatSessionKeyHash: chatSession.SessionKeyHash,
PubKey: msg.Src,
})
if err != nil {
return err
}
// confirm that client has been successfully subscribed
msgParams := whisper.MessageParams{
Dst: msg.Src,
KeySym: chatSession.SessionKey,
Topic: MakeTopic([]byte(topicAckDeviceRegistration)),
Payload: []byte(`{"server": "0x` + s.nodeID + `"}`),
TTL: uint32(s.config.TTL),
PoW: s.config.MinimumPoW,
WorkTime: 5,
}
response := whisper.NewSentMessage(&msgParams)
env, err := response.Wrap(&msgParams)
if err != nil {
return fmt.Errorf("failed to wrap server response message: %v", err)
}
if err := s.whisper.Send(env); err != nil {
return fmt.Errorf("failed to send server response message: %v", err)
}
log.Info("server confirms device registration", "dst",
common.ToHex(crypto.FromECDSAPub(msgParams.Dst)), "topic", msgParams.Topic.String())
return nil
}
// processSendNotificationRequest processes incoming client requests of type:
// when client has session key, and ready to use it to send notifications
func (s *NotificationServer) processSendNotificationRequest(msg *whisper.ReceivedMessage) error {
s.deviceSubscriptionsMu.RLock()
defer s.deviceSubscriptionsMu.RUnlock()
for _, subscriber := range s.deviceSubscriptions {
if subscriber.ChatSessionKeyHash == msg.SymKeyHash {
if whisper.IsPubKeyEqual(msg.Src, subscriber.PubKey) {
continue // no need to notify ourselves
}
if s.firebaseProvider != nil {
err := s.firebaseProvider.Send(subscriber.DeviceID, string(msg.Payload))
if err != nil {
log.Info("cannot send notification", "error", err)
}
}
}
}
return nil
}
// processClientSessionStatusRequest processes incoming client requests when:
// client wants to learn whether it is already registered on some of the servers
func (s *NotificationServer) processClientSessionStatusRequest(msg *whisper.ReceivedMessage) error {
s.clientSessionsMu.RLock()
defer s.clientSessionsMu.RUnlock()
if msg.Src == nil {
return errors.New("message 'from' field is required")
}
var sessionKey []byte
pubKey := hex.EncodeToString(crypto.FromECDSAPub(msg.Src))
for _, clientSession := range s.clientSessions {
if clientSession.ClientKey == pubKey {
sessionKey = clientSession.SessionKeyInput
break
}
}
// session is not found
if sessionKey == nil {
return nil
}
// let client know that we have session for a given public key
msgParams := whisper.MessageParams{
Src: s.protocolKey,
Dst: msg.Src,
Topic: MakeTopic([]byte(topicConfirmClientSession)),
Payload: []byte(`{"server": "0x` + s.nodeID + `", "key": "0x` + hex.EncodeToString(sessionKey) + `"}`),
TTL: uint32(s.config.TTL),
PoW: s.config.MinimumPoW,
WorkTime: 5,
}
response := whisper.NewSentMessage(&msgParams)
env, err := response.Wrap(&msgParams)
if err != nil {
return fmt.Errorf("failed to wrap server response message: %v", err)
}
if err := s.whisper.Send(env); err != nil {
return fmt.Errorf("failed to send server response message: %v", err)
}
log.Info("server confirms client session", "dst",
common.ToHex(crypto.FromECDSAPub(msgParams.Dst)), "topic", msgParams.Topic.String())
return nil
}
// processDropClientSessionRequest processes incoming client requests when:
// client wants to drop its sessions with notification servers (if they exist)
func (s *NotificationServer) processDropClientSessionRequest(msg *whisper.ReceivedMessage) error {
if msg.Src == nil {
return errors.New("message 'from' field is required")
}
s.clientSessionsMu.RLock()
pubKey := hex.EncodeToString(crypto.FromECDSAPub(msg.Src))
for _, clientSession := range s.clientSessions {
if clientSession.ClientKey == pubKey {
s.clientSessionsMu.RUnlock()
s.DropClientSession(clientSession.SessionKeyHash.Hex())
break
}
}
return nil
}
// installTopicFilter installs Whisper filter using symmetric key
func (s *NotificationServer) installTopicFilter(topicName string, topicKey []byte) (filterID string, err error) {
topic := MakeTopicAsBytes([]byte(topicName))
filter := whisper.Filter{
KeySym: topicKey,
Topics: [][]byte{topic},
AllowP2P: true,
}
filterID, err = s.whisper.Subscribe(&filter)
if err != nil {
return "", fmt.Errorf("failed installing filter: %v", err)
}
log.Debug(fmt.Sprintf("installed topic filter %v for topic %x (%s)", filterID, topic, topicName))
return
}
// installKeyFilter installs Whisper filter using asymmetric key
func (s *NotificationServer) installKeyFilter(topicName string, key *ecdsa.PrivateKey) (filterID string, err error) {
topic := MakeTopicAsBytes([]byte(topicName))
filter := whisper.Filter{
KeyAsym: key,
Topics: [][]byte{topic},
AllowP2P: true,
}
filterID, err = s.whisper.Subscribe(&filter)
if err != nil {
return "", fmt.Errorf("failed installing filter: %v", err)
}
log.Info(fmt.Sprintf("installed key filter %v for topic %x (%s)", filterID, topic, topicName))
return
}
// requestProcessorLoop processes incoming client requests, by listening to a given filter,
// and executing process function on each incoming message
func (s *NotificationServer) requestProcessorLoop(filterID string, topicWatched string, fn messageProcessingFn) {
log.Debug(fmt.Sprintf("request processor started: %s", topicWatched))
filter := s.whisper.GetFilter(filterID)
if filter == nil {
log.Warn(fmt.Sprintf("filter is not installed: %s (for topic '%s')", filterID, topicWatched))
return
}
ticker := time.NewTicker(time.Millisecond * 50)
for {
select {
case <-ticker.C:
messages := filter.Retrieve()
for _, msg := range messages {
if err := fn(msg); err != nil {
log.Warn("failed processing incoming request", "error", err)
}
}
case <-s.quit:
log.Debug("request processor stopped", "topic", topicWatched)
return
}
}
}
// makeSessionKey generates and saves random SymKey, allowing to establish secure
// channel between server and client
func (s *NotificationServer) makeSessionKey(keyName string) (sessionKey, sessionKeyDerived []byte, err error) {
// wipe out previous occurrence of symmetric key
s.whisper.DeleteSymKey(keyName)
sessionKey, err = makeSessionKey()
if err != nil {
return nil, nil, err
}
keyName, err = s.whisper.AddSymKey(keyName, sessionKey)
if err != nil {
return nil, nil, err
}
sessionKeyDerived, err = s.whisper.GetSymKey(keyName)
if err != nil {
return nil, nil, err
}
return
}

View File

@ -0,0 +1,72 @@
package notifications
import (
"crypto/sha512"
"errors"
crand "crypto/rand"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
"golang.org/x/crypto/pbkdf2"
)
// makeSessionKey returns pseudo-random symmetric key, which is used as
// session key between notification client and server
func makeSessionKey() ([]byte, error) {
// generate random key
const keyLen = 32
const size = keyLen * 2
buf := make([]byte, size)
_, err := crand.Read(buf)
if err != nil {
return nil, err
} else if !validateSymmetricKey(buf) {
return nil, errors.New("error in GenerateSymKey: crypto/rand failed to generate random data")
}
key := buf[:keyLen]
salt := buf[keyLen:]
derived, err := whisper.DeriveOneTimeKey(key, salt, whisper.EnvelopeVersion)
if err != nil {
return nil, err
} else if !validateSymmetricKey(derived) {
return nil, errors.New("failed to derive valid key")
}
return derived, nil
}
// validateSymmetricKey returns false if the key contains all zeros
func validateSymmetricKey(k []byte) bool {
return len(k) > 0 && !containsOnlyZeros(k)
}
// containsOnlyZeros checks if data is empty or not
func containsOnlyZeros(data []byte) bool {
for _, b := range data {
if b != 0 {
return false
}
}
return true
}
// MakeTopic returns Whisper topic *as bytes array* by generating cryptographic key from the provided password
func MakeTopicAsBytes(password []byte) ([]byte) {
topic := make([]byte, int(whisper.TopicLength))
x := pbkdf2.Key(password, password, 8196, 128, sha512.New)
for i := 0; i < len(x); i++ {
topic[i%whisper.TopicLength] ^= x[i]
}
return topic
}
// MakeTopic returns Whisper topic by generating cryptographic key from the provided password
func MakeTopic(password []byte) (topic whisper.TopicType) {
x := pbkdf2.Key(password, password, 8196, 128, sha512.New)
for i := 0; i < len(x); i++ {
topic[i%whisper.TopicLength] ^= x[i]
}
return
}

View File

@ -274,6 +274,11 @@ func (api *PublicWhisperAPI) Subscribe(args WhisperFilterArgs) (string, error) {
return api.whisper.Subscribe(&filter)
}
// UninstallFilter is alias for Unsubscribe
func (api *PublicWhisperAPI) UninstallFilter(id string) {
api.Unsubscribe(id)
}
// Unsubscribe disables and removes an existing filter.
func (api *PublicWhisperAPI) Unsubscribe(id string) {
api.whisper.Unsubscribe(id)

View File

@ -32,6 +32,8 @@ package whisperv5
import (
"fmt"
"time"
"github.com/ethereum/go-ethereum/p2p"
)
const (
@ -87,3 +89,15 @@ type MailServer interface {
Archive(env *Envelope)
DeliverMail(whisperPeer *Peer, request *Envelope)
}
// NotificationServer represents a notification server,
// capable of screening incoming envelopes for special
// topics, and once located, subscribe client nodes as
// recipients to notifications (push notifications atm)
type NotificationServer interface {
// Start initializes notification sending loop
Start(server *p2p.Server) error
// Stop stops notification sending loop, releasing related resources
Stop() error
}

View File

@ -72,6 +72,7 @@ type Whisper struct {
stats Statistics // Statistics of whisper node
mailServer MailServer // MailServer interface
notificationServer NotificationServer
}
// New creates a Whisper client ready to communicate through the Ethereum P2P network.
@ -119,6 +120,11 @@ func (w *Whisper) RegisterServer(server MailServer) {
w.mailServer = server
}
// RegisterNotificationServer registers notification server with Whisper
func (w *Whisper) RegisterNotificationServer(server NotificationServer) {
w.notificationServer = server
}
// Protocols returns the whisper sub-protocols ran by this particular client.
func (w *Whisper) Protocols() []p2p.Protocol {
return []p2p.Protocol{w.protocol}
@ -492,6 +498,12 @@ func (w *Whisper) Start(stack *p2p.Server) error {
go w.processQueue()
}
if w.notificationServer != nil {
if err := w.notificationServer.Start(stack); err != nil {
return err
}
}
return nil
}
@ -499,6 +511,13 @@ func (w *Whisper) Start(stack *p2p.Server) error {
// of the Whisper protocol.
func (w *Whisper) Stop() error {
close(w.quit)
if w.notificationServer != nil {
if err := w.notificationServer.Stop(); err != nil {
return err
}
}
log.Info("whisper stopped")
return nil
}