Merge pull request #17 from official-stallion/14-feat-auth

14 feat auth

README.md

@@ -4,7 +4,7 @@
 
 <p align="center">
 <img src="https://img.shields.io/badge/Golang-1.19-66ADD8?style=for-the-badge&logo=go" alt="go version" />
-<img src="https://img.shields.io/badge/Version-1.1.3-red?style=for-the-badge&logo=github" alt="version" /><br />
+<img src="https://img.shields.io/badge/Version-1.2.0-red?style=for-the-badge&logo=github" alt="version" /><br />
 <img src="https://img.shields.io/badge/MacOS-black?style=for-the-badge&logo=apple" alt="version" />
 <img src="https://img.shields.io/badge/Linux-white?style=for-the-badge&logo=linux" alt="version" />
 <img src="https://img.shields.io/badge/Windows-blue?style=for-the-badge&logo=windows" alt="version" />
@@ -13,6 +13,16 @@
 Fast message broker implemented with Golang programming language.<br />
 Using no external libraries, just internal Golang libraries.
 
+## Guide
+- [Install Stallion](#how-to-use)
+- [Setup Stallion Server](#create-server-in-golang)
+- [Using Docker](#create-a-server-with-docker)
+- [Stallion Go SDK](#creating-clients)
+  - [Subscribe](#subscribe-on-a-topic)
+  - [Publish](#publish-over-a-topic)
+  - [Unsubscribe](#unsubscribe-from-a-topic)
+- [Auth](#creating-a-server-with-auth)
+
 ## How to use?
 Get package:
 ```shell
@@ -76,3 +86,25 @@ client.Publish("topic", []byte("Hello"))
 client.Unsubscribe("topic")
 ```
 
+## Creating a server with Auth
+You can create a Stallion server with username and password for Auth.
+```go
+package main
+
+import "github.com/official-stallion/stallion"
+
+func main() {
+	if err := stallion.NewServer(":9090", "root", "Pa$$word"); err != nil {
+		panic(err)
+	}
+}
+```
+
+Now you can connect with username and password set in url.
+```go
+client, err := stallion.NewClient("st://root:Pa$$word@localhost:9090")
+if err != nil {
+    panic(err)
+}
+```
+

client.go

@@ -32,7 +32,10 @@ func NewClient(uri string) (Client, error) {
 		return nil, fmt.Errorf("failed to connect to server: %w", err)
 	}
 
-	client := internal.NewClient(conn)
+	client, err := internal.NewClient(conn, url.auth)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to server: %w", err)
+	}
 
 	return client, nil
 }

example/client/auth-client/main.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/official-stallion/stallion"
+)
+
+func main() {
+	client, err := stallion.NewClient("st://root:Pa$$word@localhost:9090")
+	if err != nil {
+		panic(err)
+	}
+
+	client.Subscribe("topic", func(data []byte) {
+		fmt.Println(string(data))
+	})
+
+	client.Publish("topic", []byte("Hello"))
+
+	time.Sleep(3 * time.Second)
+}

example/server/auth-server/main.go

@@ -0,0 +1,9 @@
+package main
+
+import "github.com/official-stallion/stallion"
+
+func main() {
+	if err := stallion.NewServer(":9090", "root", "Pa$$word"); err != nil {
+		panic(err)
+	}
+}

internal/client.go

@@ -1,6 +1,7 @@
 package internal
 
 import (
+	"fmt"
 	"net"
 	"time"
 )
@@ -28,7 +29,7 @@ type client struct {
 }
 
 // NewClient creates a new client handler.
-func NewClient(conn net.Conn) *client {
+func NewClient(conn net.Conn, auth string) (*client, error) {
 	c := &client{
 		topics:             make(map[string]MessageHandler),
 		communicateChannel: make(chan message),
@@ -38,13 +39,18 @@ func NewClient(conn net.Conn) *client {
 		},
 	}
 
+	// send the ping message
+	if err := c.ping([]byte(auth)); err != nil {
+		return nil, fmt.Errorf("failed to create client: %w", err)
+	}
+
 	// starting data reader
 	go c.readDataFromServer()
 
 	// start listening on channels
 	go c.listen()
 
-	return c
+	return c, nil
 }
 
 // readDataFromServer gets all data from server.
@@ -98,6 +104,38 @@ func (c *client) close() {
 	_ = c.network.connection.Close()
 }
 
+// send a ping message to stallion server.
+func (c *client) ping(data []byte) error {
+	// sending ping data as a message
+	if err := c.network.send(encodeMessage(newMessage(PingMessage, "", data))); err != nil {
+		return fmt.Errorf("failed to ping server: %w", err)
+	}
+
+	// creating a buffer
+	var buffer = make([]byte, 2048)
+
+	// read data from network
+	tmp, er := c.network.get(buffer)
+	if er != nil {
+		return fmt.Errorf("server failed to pong: %w", er)
+	}
+
+	// check for response
+	response, err := decodeMessage(tmp)
+	if err != nil {
+		return fmt.Errorf("decode message failed")
+	}
+
+	switch response.Type {
+	case PongMessage:
+		return nil
+	case Imposter:
+		return fmt.Errorf("unauthorized user")
+	default:
+		return fmt.Errorf("connection failed")
+	}
+}
+
 // Publish will send a message to broker server.
 func (c *client) Publish(topic string, data []byte) error {
 	err := c.network.send(encodeMessage(newMessage(Text, topic, data)))

internal/enum.go

@@ -5,4 +5,7 @@ const (
 	Text        int = iota + 1 // normal message
 	Subscribe                  // subscribe message
 	Unsubscribe                // unsubscribe message
+	PingMessage                // ping message
+	PongMessage                // pong message
+	Imposter                   // unauthorized user message
 )

internal/server.go

@@ -7,13 +7,18 @@ import (
 
 // server is our broker service.
 type server struct {
+	user string
+	pass string
+
 	prefix int
 	broker *broker
 }
 
 // NewServer returns a new broker server.
-func NewServer() *server {
+func NewServer(user string, pass string) *server {
 	s := &server{
+		user:   user,
+		pass:   pass,
 		prefix: 101,
 	}
 
@@ -33,6 +38,8 @@ func NewServer() *server {
 func (s *server) Handle(conn net.Conn) {
 	w := newWorker(
 		s.prefix,
+		s.user,
+		s.pass,
 		conn,
 		make(chan message),
 		s.broker.receiveChannel,

internal/types.go

@@ -3,20 +3,20 @@ package internal
 // MessageHandler is a handler for messages that come from subscribing.
 type MessageHandler func([]byte)
 
-// WorkerChannel is worker channel with its id.
+// workerChannel is worker channel with its id.
 type workerChannel struct {
 	id      int
 	channel chan message
 }
 
-// SubscribeChannel is for subscribe data channel.
+// subscribeChannel is for subscribe data channel.
 type subscribeChannel struct {
 	id      int
 	topic   string
 	channel chan message
 }
 
-// UnsubscribeChannel is for unsubscribe data channel.
+// unsubscribeChannel is for unsubscribe data channel.
 type unsubscribeChannel struct {
 	id    int
 	topic string

internal/worker.go

@@ -1,7 +1,9 @@
 package internal
 
 import (
+	"fmt"
 	"net"
+	"strings"
 	"time"
 )
 
@@ -11,6 +13,12 @@ type worker struct {
 	// each worker has its unique id
 	id int
 
+	// authentication fields
+	// user of stallion client
+	user string
+	// pass of stallion client
+	pass string
+
 	// for network socket handling
 	network network
 
@@ -33,14 +41,18 @@ type worker struct {
 // newWorker generates a new worker.
 func newWorker(
 	id int,
+	user string,
+	pass string,
 	conn net.Conn,
 	sen, rec chan message,
 	sub chan subscribeChannel,
 	unsub chan unsubscribeChannel,
 	ter chan int,
 ) *worker {
 	return &worker{
-		id: id,
+		id:   id,
+		user: user,
+		pass: pass,
 		network: network{
 			connection: conn,
 		},
@@ -57,6 +69,15 @@ func (w *worker) start() {
 	// closing channel after we are done
 	defer close(w.sendChannel)
 
+	// check the ping pong connection
+	if err := w.pong(); err != nil {
+		logError("failed to pong client", err)
+
+		w.terminateChannel <- w.id
+
+		return
+	}
+
 	// start for input data
 	go w.arrival()
 
@@ -68,6 +89,43 @@ func (w *worker) start() {
 	}
 }
 
+// get client ping message.
+func (w *worker) pong() error {
+	// creating a buffer
+	var buffer = make([]byte, 2048)
+
+	// read data from network
+	tmp, er := w.network.get(buffer)
+	if er != nil {
+		return fmt.Errorf("client failed to ping: %w", er)
+	}
+
+	// get user request
+	request, err := decodeMessage(tmp)
+	if err != nil {
+		return fmt.Errorf("decode message failed")
+	}
+
+	data := strings.Split(string(request.Data), ":")
+
+	// check auth
+	if w.user == data[0] && w.pass == data[1] {
+		// send pong response
+		if e := w.network.send(encodeMessage(newMessage(PongMessage, "", nil))); e != nil {
+			return fmt.Errorf("failed to pong client: %w", e)
+		}
+
+		return nil
+	}
+
+	// return sabotage message
+	if e := w.network.send(encodeMessage(newMessage(Imposter, "", nil))); e != nil {
+		return fmt.Errorf("failed to pong client: %w", e)
+	}
+
+	return fmt.Errorf("un-auth client")
+}
+
 // transfer will send a data byte through handler.
 func (w *worker) transfer(data message) {
 	err := w.network.send(encodeMessage(data))

server.go

@@ -13,9 +13,24 @@ type Server interface {
 }
 
 // NewServer creates a new broker server on given port.
-func NewServer(port string) error {
+func NewServer(port string, auth ...string) error {
+	// get authentication options
+	var (
+		user string
+		pass string
+	)
+
+	// setting the authentication options
+	if len(auth) > 1 {
+		user = auth[0]
+		pass = auth[1]
+	} else {
+		user = " "
+		pass = " "
+	}
+
 	// creating a new server
-	serve := internal.NewServer()
+	serve := internal.NewServer(user, pass)
 
 	// listen over a port
 	listener, err := net.Listen("tcp", port)