Decouple request and response via WebSocket
parent
08be6e68bc
commit
2093bf2429
@ -0,0 +1,172 @@
|
|||||||
|
package handlers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"crypto/x509"
|
||||||
|
"encoding/pem"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io/ioutil"
|
||||||
|
"os"
|
||||||
|
"os/exec"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
|
log "github.com/sirupsen/logrus"
|
||||||
|
)
|
||||||
|
|
||||||
|
type SigningRequestRegistry struct {
|
||||||
|
caCertificates []*x509.Certificate
|
||||||
|
caChainMap map[string][]string
|
||||||
|
requests map[string]chan *responseData
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewSigningRequestRegistry(caCertificates []*x509.Certificate) *SigningRequestRegistry {
|
||||||
|
return &SigningRequestRegistry{
|
||||||
|
caCertificates: caCertificates,
|
||||||
|
caChainMap: make(map[string][]string),
|
||||||
|
requests: make(map[string]chan *responseData),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (registry *SigningRequestRegistry) AddSigningRequest(request *requestData) (string, error) {
|
||||||
|
requestUuid, err := uuid.NewRandom()
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
go func() {
|
||||||
|
responseChannel := make(chan *responseData, 1)
|
||||||
|
registry.requests[requestUuid.String()] = responseChannel
|
||||||
|
registry.signCertificate(responseChannel, request)
|
||||||
|
}()
|
||||||
|
return requestUuid.String(), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (registry *SigningRequestRegistry) signCertificate(channel chan *responseData, request *requestData) {
|
||||||
|
responseData, err := registry.sign(request)
|
||||||
|
if err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
close(channel)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
channel <- responseData
|
||||||
|
}
|
||||||
|
|
||||||
|
func (registry *SigningRequestRegistry) sign(request *requestData) (response *responseData, err error) {
|
||||||
|
log.Debugf("received CSR for %s:\n\n%s", request.CommonName, request.Csr)
|
||||||
|
subjectDN := fmt.Sprintf("/CN=%s", request.CommonName)
|
||||||
|
var csrFile *os.File
|
||||||
|
if csrFile, err = ioutil.TempFile("", "*.csr.pem"); err != nil {
|
||||||
|
log.Errorf("could not open temporary file: %s", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if _, err = csrFile.Write([]byte(request.Csr)); err != nil {
|
||||||
|
log.Errorf("could not write CSR to file: %s", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err = csrFile.Close(); err != nil {
|
||||||
|
log.Errorf("could not close CSR file: %s", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer func(file *os.File) {
|
||||||
|
err = os.Remove(file.Name())
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("could not remove temporary file: %s", err)
|
||||||
|
}
|
||||||
|
}(csrFile)
|
||||||
|
|
||||||
|
// simulate a delay during certificate creation
|
||||||
|
time.Sleep(5 * time.Second)
|
||||||
|
|
||||||
|
opensslCommand := exec.Command(
|
||||||
|
"openssl", "ca", "-config", "ca.cnf",
|
||||||
|
"-policy", "policy_match", "-extensions", "client_ext",
|
||||||
|
"-batch", "-subj", subjectDN, "-utf8", "-rand_serial", "-in", "in.pem")
|
||||||
|
var out, cmdErr bytes.Buffer
|
||||||
|
opensslCommand.Stdout = &out
|
||||||
|
opensslCommand.Stderr = &cmdErr
|
||||||
|
err = opensslCommand.Run()
|
||||||
|
if err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
log.Error(cmdErr.String())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var block *pem.Block
|
||||||
|
if block, _ = pem.Decode(out.Bytes()); block == nil {
|
||||||
|
err = fmt.Errorf("could not decode pem")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var certificate *x509.Certificate
|
||||||
|
if certificate, err = x509.ParseCertificate(block.Bytes); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var caChain []string
|
||||||
|
if caChain, err = registry.getCAChain(certificate); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
response = &responseData{
|
||||||
|
Certificate: string(pem.EncodeToMemory(&pem.Block{
|
||||||
|
Type: "CERTIFICATE",
|
||||||
|
Bytes: certificate.Raw,
|
||||||
|
})),
|
||||||
|
CAChain: caChain,
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
func (registry *SigningRequestRegistry) GetResponseChannel(requestUuid string) (chan *responseData, error) {
|
||||||
|
if responseChannel, exists := registry.requests[requestUuid]; exists {
|
||||||
|
delete(registry.requests, requestUuid)
|
||||||
|
return responseChannel, nil
|
||||||
|
} else {
|
||||||
|
return nil, errors.New("no request found")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (registry *SigningRequestRegistry) getCAChain(certificate *x509.Certificate) ([]string, error) {
|
||||||
|
issuerString := string(certificate.RawIssuer)
|
||||||
|
|
||||||
|
if value, exists := registry.caChainMap[issuerString]; exists {
|
||||||
|
return value, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
result := make([]string, 0)
|
||||||
|
|
||||||
|
appendCert := func(cert *x509.Certificate) {
|
||||||
|
result = append(
|
||||||
|
result,
|
||||||
|
string(pem.EncodeToMemory(&pem.Block{Bytes: cert.Raw, Type: "CERTIFICATE"})))
|
||||||
|
log.Debugf("added %s to cachain", result[len(result)-1])
|
||||||
|
}
|
||||||
|
|
||||||
|
var previous *x509.Certificate
|
||||||
|
var count = 0
|
||||||
|
for {
|
||||||
|
if len(registry.caCertificates) == 0 {
|
||||||
|
return nil, errors.New("no CA certificates loaded")
|
||||||
|
}
|
||||||
|
if count > len(registry.caCertificates) {
|
||||||
|
return nil, errors.New("could not construct certificate chain")
|
||||||
|
}
|
||||||
|
for _, caCert := range registry.caCertificates {
|
||||||
|
if previous == nil {
|
||||||
|
if bytes.Equal(caCert.RawSubject, certificate.RawIssuer) {
|
||||||
|
previous = caCert
|
||||||
|
appendCert(caCert)
|
||||||
|
}
|
||||||
|
} else if bytes.Equal(previous.RawSubject, previous.RawIssuer) {
|
||||||
|
registry.caChainMap[issuerString] = result
|
||||||
|
return result, nil
|
||||||
|
} else if bytes.Equal(caCert.RawSubject, previous.RawIssuer) {
|
||||||
|
previous = caCert
|
||||||
|
appendCert(caCert)
|
||||||
|
} else {
|
||||||
|
log.Debugf("skipped certificate %s", caCert.Subject)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
count++
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,79 @@
|
|||||||
|
package handlers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"net/http"
|
||||||
|
|
||||||
|
"github.com/gobwas/ws"
|
||||||
|
"github.com/gobwas/ws/wsutil"
|
||||||
|
log "github.com/sirupsen/logrus"
|
||||||
|
)
|
||||||
|
|
||||||
|
type WebSocketHandler struct {
|
||||||
|
requestRegistry *SigningRequestRegistry
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewWebSocketHandler(registry *SigningRequestRegistry) *WebSocketHandler {
|
||||||
|
return &WebSocketHandler{requestRegistry: registry}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w *WebSocketHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
|
||||||
|
conn, _, _, err := ws.UpgradeHTTP(request, writer)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(writer, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
defer func() { _ = conn.Close() }()
|
||||||
|
|
||||||
|
var (
|
||||||
|
reader = wsutil.NewReader(conn, ws.StateServerSide)
|
||||||
|
writer = wsutil.NewWriter(conn, ws.StateServerSide, ws.OpText)
|
||||||
|
jsonDecoder = json.NewDecoder(reader)
|
||||||
|
jsonEncoder = json.NewEncoder(writer)
|
||||||
|
)
|
||||||
|
|
||||||
|
for {
|
||||||
|
header, err := reader.NextFrame()
|
||||||
|
if err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
if header.OpCode == ws.OpClose {
|
||||||
|
log.Debug("channel closed")
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
type requestType struct {
|
||||||
|
RequestId string `json:"request_id"`
|
||||||
|
}
|
||||||
|
|
||||||
|
request := &requestType{}
|
||||||
|
err = jsonDecoder.Decode(request)
|
||||||
|
if err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
channel, err := w.requestRegistry.GetResponseChannel(request.RequestId)
|
||||||
|
if err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
var response *responseData
|
||||||
|
response = <-channel
|
||||||
|
if err = jsonEncoder.Encode(response); err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
close(channel)
|
||||||
|
|
||||||
|
if err = writer.Flush(); err != nil {
|
||||||
|
log.Error(err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
Loading…
Reference in New Issue