probe_gateway_rpc/servlet/auth-servlet.go

package servlet

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io/ioutil"

	"github.com/valyala/fasthttp"

	"git.loafle.net/commons/logging-go"
	"git.loafle.net/commons/server-go"
	"git.loafle.net/commons/server-go/socket"
	occn "git.loafle.net/overflow/commons-go/config/noauthprobe"
	ocmm "git.loafle.net/overflow/commons-go/model/meta"
	ocmn "git.loafle.net/overflow/commons-go/model/noauthprobe"
	oe "git.loafle.net/overflow/external-go"
	oeg "git.loafle.net/overflow/external-go/grpc"
	og "git.loafle.net/overflow/gateway"
	ogs "git.loafle.net/overflow/gateway/subscribe"
	ogrs "git.loafle.net/overflow/gateway_rpc/servlet"
	opgrs "git.loafle.net/overflow/probe_gateway_rpc/subscribe"
)

type AuthServlet interface {
	ogrs.RPCServlet
}

type AuthServlets struct {
	ogrs.RPCServlets
}

func (s *AuthServlets) Init(serverCtx server.ServerCtx) error {
	if err := s.RPCServlets.Init(serverCtx); nil != err {
		return err
	}

	return nil
}

func (s *AuthServlets) OnStart(serverCtx server.ServerCtx) error {
	if err := s.RPCServlets.OnStart(serverCtx); nil != err {
		return err
	}

	subscribeChan, err := opgrs.Subscriber.Subscribe(occn.HTTPEntry_Auth)
	if nil != err {
		return err
	}
	go s.handleSubscribe(serverCtx, subscribeChan)

	return nil
}

func (s *AuthServlets) OnStop(serverCtx server.ServerCtx) {
	if err := opgrs.Subscriber.Unsubscribe(occn.HTTPEntry_Auth); nil != err {
		logging.Logger().Warn(err)
	}

	s.RPCServlets.OnStop(serverCtx)
}

func (s *AuthServlets) Destroy(serverCtx server.ServerCtx) {

	s.RPCServlets.Destroy(serverCtx)
}

func (s *AuthServlets) Handshake(servletCtx server.ServletCtx, ctx *fasthttp.RequestCtx) (*fasthttp.ResponseHeader, error) {
	bMethod := ctx.Request.Header.Peek(occn.HTTPRequestHeaderKey_NoAuthProbe_Method)

	if nil == bMethod {
		return nil, fmt.Errorf("Unexpected noauth probe method: %v", bMethod)
	}

	method := string(bMethod)

	switch method {
	case occn.HTTPRequestHeaderValue_NoAuthProbe_Method_Regist:
		bInfo := ctx.Request.Header.Peek(occn.HTTPRequestHeaderKey_NoAuthProbe_Info)
		if nil == bInfo {
			return nil, fmt.Errorf("Unexpected Request Header Key NoAuthProbe Info")
		}

		buf := make([]byte, base64.StdEncoding.DecodedLen(len(bInfo)))
		_, err := base64.StdEncoding.Decode(buf, bInfo)
		if nil != err {
			return nil, fmt.Errorf("Base64 Encoding Error: %s", err.Error())
		}

		gReader, err := gzip.NewReader(bytes.NewBuffer(buf))
		if nil != err {
			return nil, fmt.Errorf("gzip Error: %s", err.Error())
		}
		defer gReader.Close()

		rJSON, err := ioutil.ReadAll(gReader)
		if nil != err {
			return nil, fmt.Errorf("gzip Error: %s", err.Error())
		}

		grpcCTX := context.Background()
		r, err := oeg.Exec(grpcCTX, "NoAuthProbeService.regist", string(rJSON))
		if nil != err {
			return nil, fmt.Errorf("grpc call Error: %s", err.Error())
		}

		nap := &ocmn.NoAuthProbe{}
		err = json.Unmarshal([]byte(r), nap)
		if nil != err {
			return nil, fmt.Errorf("grpc result unMarshal Error: %s", err.Error())
		}

		extHeader := &fasthttp.ResponseHeader{}
		extHeader.Add(occn.HTTPResponseHeaderKey_NoAuthProbe_Method, occn.HTTPResponseHeaderValue_NoAuthProbe_Method_TempProbeKey)
		extHeader.Add(occn.HTTPResponseHeaderKey_NoAuthProbe_Method_Param, nap.TempProbeKey)

		servletCtx.SetAttribute(og.SessionIDKey, nap.TempProbeKey)
		servletCtx.SetAttribute(og.SessionClientTypeKey, oe.PROBE)
		servletCtx.SetAttribute(og.SessionTargetIDKey, nap.TempProbeKey)

		return extHeader, nil
	case occn.HTTPRequestHeaderValue_NoAuthProbe_Method_Connect:
		bTempProbeKey := ctx.Request.Header.Peek(occn.HTTPRequestHeaderKey_NoAuthProbe_TempProbeKey)
		if nil == bTempProbeKey {
			return nil, fmt.Errorf("Noauth probe temp key is not an existing key")
		}
		tempProbeKey := string(bTempProbeKey)

		grpcCTX := context.Background()
		r, err := oeg.Exec(grpcCTX, "NoAuthProbeService.readByTempProbeKey", tempProbeKey)
		if nil != err {
			return nil, fmt.Errorf("grpc result error: %s", err.Error())
		}

		nap := &ocmn.NoAuthProbe{}
		err = json.Unmarshal([]byte(r), nap)
		if nil != err {
			return nil, fmt.Errorf("grpc result unMarshal Error: %s", err.Error())
		}

		extHeader := &fasthttp.ResponseHeader{}

		switch ocmm.ToMetaNoAuthProbeStatusEnum(nap.MetaNoAuthProbeStatus) {
		case ocmm.MetaNoAuthProbeStatusEnumACCEPT:
			extHeader.Add(occn.HTTPResponseHeaderKey_NoAuthProbe_Method, occn.HTTPResponseHeaderValue_NoAuthProbe_Method_Accept)
			extHeader.Add(occn.HTTPResponseHeaderKey_NoAuthProbe_Method_Param, nap.Probe.ProbeKey)

		case ocmm.MetaNoAuthProbeStatusEnumDENY:
			extHeader.Add(occn.HTTPResponseHeaderKey_NoAuthProbe_Method, occn.HTTPResponseHeaderValue_NoAuthProbe_Method_Deny)

		default:
		}

		servletCtx.SetAttribute(og.SessionIDKey, tempProbeKey)
		servletCtx.SetAttribute(og.SessionClientTypeKey, oe.PROBE)
		servletCtx.SetAttribute(og.SessionTargetIDKey, tempProbeKey)

		return extHeader, nil

	default:
		return nil, fmt.Errorf("Unexpected noauth probe method: %s", method)
	}
}

func (s *AuthServlets) OnConnect(servletCtx server.ServletCtx, conn socket.Conn) {
	s.RPCServlets.OnConnect(servletCtx, conn)

	_tempProbeKey := servletCtx.GetAttribute(og.SessionIDKey)
	if nil != _tempProbeKey {
		grpcCTX := context.Background()
		_, err := oeg.Exec(grpcCTX, "NoAuthProbeService.onConnect", _tempProbeKey.(string), conn.RemoteAddr().String())
		if nil != err {
			logging.Logger().Errorf("onConnect error %v", err)
		}
	}
}

func (s *AuthServlets) OnDisconnect(servletCtx server.ServletCtx) {
	s.RPCServlets.OnDisconnect(servletCtx)

	_tempProbeKey := servletCtx.GetAttribute(og.SessionIDKey)
	if nil != _tempProbeKey {
		grpcCTX := context.Background()
		_, err := oeg.Exec(grpcCTX, "NoAuthProbeService.onDisconnect", _tempProbeKey.(string))
		if nil != err {
			logging.Logger().Errorf("onDisconnect error %v", err)
		}
	}
}

func (s *AuthServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeChan <-chan *ogs.Message) {
	var sessions []*ogrs.Session

LOOP:
	for {
		select {
		case msg, ok := <-subscribeChan:
			if !ok {
				return
			}

			switch msg.TargetType {
			case ogs.PROBE:
				sessions = s.GetSessions(msg.Targets)
			default:
				logging.Logger().Warnf("Subscriber: Unknown TargetType %s", msg.TargetType)
				continue LOOP
			}
			if nil == sessions || 0 == len(sessions) {
				continue LOOP
			}

			for _, session := range sessions {
				_writeChan := session.ServletCtx.GetAttribute(og.SessionWriteChanKey)
				if nil != _writeChan {
					writeChan := _writeChan.(chan<- []byte)
					writeChan <- *msg.Message
				}
			}
		}
	}
}