watcher

README

Casbin Go Cloud Development kit based watcher

Casbin watcher built on top of gocloud.dev.

Installation

go get github.com/rusenask/casbin-go-cloud-watcher

Usage

Configuration is slightly different for each provider as it needs to get different settings from environment. You can read more about URLs and configuration here: https://gocloud.dev/concepts/urls/.

Supported providers:

• NATS
• GCP Cloud Pub/Sub
• AWS SNS
• AWS SQS
• Azure Service Bus
• Apache Kafka
• RabbitMQ
• In memory pubsub (useful for local testing and single node installs)

You can view provider configuration examples here: https://github.com/google/go-cloud/tree/master/pubsub.

NATS

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"
    // Enable NATS driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/natspubsub"
    
    "github.com/casbin/casbin"
)

func main() {
    // This URL will Dial the NATS server at the URL in the environment variable
	  // NATS_SERVER_URL and send messages with subject "casbin-policy-updates".
    os.Setenv("NATS_SERVER_URL", "nats://localhost:4222")
    ctx, cancel := context.WithCancel(context.Background())
	  defer cancel()
    watcher, _ := cloudwatcher.New(ctx, "nats://casbin-policy-updates")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

In Memory

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"
    // Enable in-memory driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/mempubsub"
    
    "github.com/casbin/casbin"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
	  defer cancel()
    watcher, _ := cloudwatcher.New(ctx, "mem://topicA")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

GCP Cloud Pub/Sub

URLs are gcppubsub://projects/myproject/topics/mytopic. The URLs use the project ID and the topic ID. See Application Default Credentials to learn about authentication alternatives, including using environment variables.

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"
    // Enable in GCP pubsub driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/gcppubsub"

    "github.com/casbin/casbin"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    watcher, _ := cloudwatcher.New(ctx, "gcppubsub://projects/myproject/topics/mytopic")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

Amazon Simple Notification Service (SNS)

Watcher can publish to an Amazon Simple Notification Service (SNS) topic. SNS URLs in the Go CDK use the Amazon Resource Name (ARN) to identify the topic. You should specify the region query parameter to ensure your application connects to the correct region.

Watcher will create a default AWS Session with the SharedConfigEnable option enabled; if you have authenticated with the AWS CLI, it will use those credentials. See AWS Session to learn about authentication alternatives, including using environment variables.

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"
    // Enable AWS SQS & SNS driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/awssnssqs"


    "github.com/casbin/casbin"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    const topicARN = "arn:aws:sns:us-east-2:123456789012:mytopic"
    // Note the 3 slashes; ARNs have multiple colons and therefore aren't valid
    // as hostnames in the URL.
    watcher, _ := cloudwatcher.New(ctx, "awssns:///"+topicARN+"?region=us-east-2")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

Amazon Simple Queue Service (SQS)

Watcher can publish to an Amazon Simple Queue Service (SQS) topic. SQS URLs closely resemble the the queue URL, except the leading https:// is replaced with awssqs://. You can specify the region query parameter to ensure your application connects to the correct region, but otherwise watcher will use the region found in the environment variables or your AWS CLI configuration.

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"/
    // Enable AWS SQS & SNS driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/awssnssqs"
    "github.com/casbin/casbin"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    // https://docs.aws.amazon.com/sdk-for-net/v2/developer-guide/QueueURL.html
    const queueURL = "https://sqs.us-east-2.amazonaws.com/123456789012/myqueue"
    watcher, _ := cloudwatcher.New(ctx, "awssqs://"+queueURL+"?region=us-east-2")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

Azure Service Bus

Watcher can publish to an Azure Service Bus topic over AMQP 1.0. The URL for publishing is the topic name. pubsub.OpenTopic will use the environment variable SERVICEBUS_CONNECTION_STRING to obtain the Service Bus connection string. The connection string can be obtained from the Azure portal.

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"
    // Enable Azure Service Bus driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/azuresb"

    "github.com/casbin/casbin"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()    
    
    // Watcher creates a *pubsub.Topic from a URL.
    // This URL will open the topic "mytopic" using a connection string
    // from the environment variable SERVICEBUS_CONNECTION_STRING.
    const queueURL = "https://sqs.us-east-2.amazonaws.com/123456789012/myqueue"
    watcher, _ := cloudwatcher.New(ctx, "azuresb://mytopic")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

Kafka

Watcher can publish to a Kafka cluster. A Kafka URL only includes the topic name. The brokers in the Kafka cluster are discovered from the KAFKA_BROKERS environment variable (which is a comma-delimited list of hosts, something like 1.2.3.4:9092,5.6.7.8:9092).

import (
    cloudwatcher "github.com/rusenask/casbin-go-cloud-watcher"
    // Enable Kafka driver
    _ "github.com/rusenask/casbin-go-cloud-watcher/drivers/kafkapubsub"

    "github.com/casbin/casbin"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()    
    
    // Watcher creates a *pubsub.Topic from a URL.
    // The host + path are the topic name to send to.
    // The set of brokers must be in an environment variable KAFKA_BROKERS.
    const queueURL = "https://sqs.us-east-2.amazonaws.com/123456789012/myqueue"
    watcher, _ := cloudwatcher.New(ctx, "kafka://my-topic")

    enforcer := casbin.NewSyncedEnforcer("model.conf", "policy.csv")
    enforcer.SetWatcher(watcher)

    watcher.SetUpdateCallback(func(m string) {
      enforcer.LoadPolicy()
    })
}

About Go Cloud Dev

Portable Cloud APIs in Go. Strives to implement these APIs for the leading Cloud providers: AWS, GCP and Azure, as well as provide a local (on-prem) implementation such as Kafka, NATS, etc.

Using the Go CDK you can write your application code once using these idiomatic APIs, test locally using the local versions, and then deploy to a cloud provider with only minimal setup-time changes.

Documentation

Index

• Variables
• type Watcher
  • func New(ctx context.Context, url ...string) (*Watcher, error)
  • func (w *Watcher) Close()
  • func (w *Watcher) SetUpdateCallback(callbackFunc func(string)) error
  • func (w *Watcher) Update() error

Variables

var (
	ErrNotConnected = errors.New("pubsub not connected, cannot dispatch update message")
)

Errors

Types

type Watcher

type Watcher struct {
	// contains filtered or unexported fields
}

Watcher implements Casbin updates watcher to synchronize policy changes between the nodes

func New

func New(ctx context.Context, url ...string) (*Watcher, error)

New creates a new watcher https://gocloud.dev/concepts/urls/ gcppubsub://myproject/mytopic

func (*Watcher) Close

func (w *Watcher) Close()

Close stops and releases the watcher, the callback function will not be called any more.

func (*Watcher) SetUpdateCallback

func (w *Watcher) SetUpdateCallback(callbackFunc func(string)) error

SetUpdateCallback sets the callback function that the watcher will call when the policy in DB has been changed by other instances. A classic callback is Enforcer.LoadPolicy().

func (*Watcher) Update

func (w *Watcher) Update() error

Update calls the update callback of other instances to synchronize their policy. It is usually called after changing the policy in DB, like Enforcer.SavePolicy(), Enforcer.AddPolicy(), Enforcer.RemovePolicy(), etc.