406 lines
13 KiB
Go
406 lines
13 KiB
Go
package admin
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"proxy-pool/internal/domain/adminstate"
|
|
"proxy-pool/internal/platform/httpapi"
|
|
"proxy-pool/internal/platform/httpsecurity"
|
|
)
|
|
|
|
const (
|
|
statusPath = "/api/v1/status"
|
|
auditPath = "/api/v1/audit"
|
|
reloadPath = "/api/v1/config/reload"
|
|
upstreamPrefix = "/api/v1/upstreams/"
|
|
routingPrefix = "/api/v1/routing/"
|
|
maxResourceNameBytes = 128
|
|
defaultAuditPageSize = 100
|
|
)
|
|
|
|
var (
|
|
ErrInvalidHandler = errors.New("invalid admin HTTP handler")
|
|
ErrNotFound = errors.New("admin resource not found")
|
|
ErrConflict = errors.New("admin mutation conflict")
|
|
ErrInvalidConfiguration = errors.New("invalid configuration")
|
|
ErrUnavailable = errors.New("admin service unavailable")
|
|
)
|
|
|
|
type Service interface {
|
|
Status(context.Context) (Status, error)
|
|
ReadAudit(context.Context, AuditQuery) (AuditPage, error)
|
|
SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error)
|
|
SwitchRouting(context.Context, SwitchCommand) (MutationResult, error)
|
|
ReloadConfiguration(context.Context, ReloadCommand) (MutationResult, error)
|
|
}
|
|
|
|
type IdentityResolver interface {
|
|
Resolve(*http.Request) (httpsecurity.Identity, error)
|
|
}
|
|
|
|
var _ IdentityResolver = (*httpsecurity.Protection)(nil)
|
|
|
|
type Options struct {
|
|
MaxBodyBytes int64
|
|
}
|
|
|
|
type Status struct {
|
|
ConfigVersion string `json:"configVersion"`
|
|
SnapshotVersion uint64 `json:"snapshotVersion"`
|
|
Upstreams []UpstreamStatus `json:"upstreams"`
|
|
Workers []WorkerStatus `json:"workers"`
|
|
}
|
|
|
|
type UpstreamStatus struct {
|
|
Name string `json:"name"`
|
|
Enabled bool `json:"enabled"`
|
|
Available int64 `json:"available"`
|
|
Checking int64 `json:"checking"`
|
|
Suspect int64 `json:"suspect"`
|
|
Draining int64 `json:"draining"`
|
|
Extracted int64 `json:"extracted"`
|
|
ConsecutiveEmptyFetch int64 `json:"consecutiveEmptyFetch,omitempty"`
|
|
FetchErrorCount int64 `json:"fetchErrorCount,omitempty"`
|
|
}
|
|
|
|
type WorkerStatus struct {
|
|
ID string `json:"id"`
|
|
Zone string `json:"zone"`
|
|
Connected bool `json:"connected"`
|
|
SnapshotVersion uint64 `json:"snapshotVersion"`
|
|
StaleSeconds int64 `json:"staleSeconds,omitempty"`
|
|
}
|
|
|
|
type MutationResult struct {
|
|
RequestID string `json:"requestId"`
|
|
Changed bool `json:"changed"`
|
|
Version uint64 `json:"version"`
|
|
Message string `json:"message,omitempty"`
|
|
}
|
|
|
|
type AuditQuery struct {
|
|
AfterID uint64
|
|
Limit int
|
|
}
|
|
|
|
func (query AuditQuery) Validate() error {
|
|
if query.Limit <= 0 || query.Limit > adminstate.MaxPageSize {
|
|
return ErrInvalidConfiguration
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type AuditRecord struct {
|
|
ID uint64 `json:"id"`
|
|
RequestID string `json:"requestId"`
|
|
ActorID string `json:"actorId"`
|
|
SourceIP string `json:"sourceIp,omitempty"`
|
|
Action string `json:"action"`
|
|
ResourceType string `json:"resourceType"`
|
|
ResourceName string `json:"resourceName"`
|
|
Changed bool `json:"changed"`
|
|
Version uint64 `json:"version"`
|
|
Reason string `json:"reason,omitempty"`
|
|
OccurredAt time.Time `json:"occurredAt"`
|
|
}
|
|
|
|
type AuditPage struct {
|
|
Records []AuditRecord `json:"records"`
|
|
}
|
|
|
|
type SetUpstreamCommand struct {
|
|
RequestID string
|
|
ActorID string
|
|
SourceIP string
|
|
Name string
|
|
Enabled bool
|
|
}
|
|
|
|
type SwitchCommand struct {
|
|
RequestID string `json:"-"`
|
|
ActorID string `json:"-"`
|
|
SourceIP string `json:"-"`
|
|
Name string `json:"-"`
|
|
ExpectedCurrent string `json:"expectedCurrent"`
|
|
Target string `json:"target"`
|
|
Reason string `json:"reason,omitempty"`
|
|
}
|
|
|
|
type ReloadCommand struct {
|
|
RequestID string
|
|
ActorID string
|
|
SourceIP string
|
|
}
|
|
|
|
type Handler struct {
|
|
service Service
|
|
identity IdentityResolver
|
|
maxBodyBytes int64
|
|
}
|
|
|
|
func NewHandler(service Service, identity IdentityResolver, options Options) (*Handler, error) {
|
|
if service == nil || identity == nil || options.MaxBodyBytes <= 0 {
|
|
return nil, ErrInvalidHandler
|
|
}
|
|
return &Handler{service: service, identity: identity, maxBodyBytes: options.MaxBodyBytes}, nil
|
|
}
|
|
|
|
func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
|
|
requestID, err := httpapi.ResolveRequestID(request)
|
|
if err != nil {
|
|
writeTransportProblem(writer, http.StatusBadRequest, "INVALID_REQUEST_ID", "Invalid request ID", "X-Request-ID is invalid", requestID)
|
|
return
|
|
}
|
|
identity, err := handler.identity.Resolve(request)
|
|
if err != nil {
|
|
if !httpsecurity.WriteProblem(writer, requestID, err) {
|
|
writeTransportProblem(writer, http.StatusInternalServerError, "INTERNAL_ERROR", "Internal server error", "the request could not be completed", requestID)
|
|
}
|
|
return
|
|
}
|
|
|
|
switch request.URL.Path {
|
|
case statusPath:
|
|
if !requireMethod(writer, request, http.MethodGet, requestID) {
|
|
return
|
|
}
|
|
handler.getStatus(writer, request, requestID)
|
|
return
|
|
case auditPath:
|
|
if !requireMethod(writer, request, http.MethodGet, requestID) {
|
|
return
|
|
}
|
|
handler.readAudit(writer, request, requestID)
|
|
return
|
|
case reloadPath:
|
|
if !requireMethod(writer, request, http.MethodPost, requestID) {
|
|
return
|
|
}
|
|
handler.reload(writer, request, requestID, identity)
|
|
return
|
|
}
|
|
|
|
if name, action, ok := matchNamedAction(request.URL.Path, upstreamPrefix, "enable", "disable"); ok {
|
|
if !requireMethod(writer, request, http.MethodPost, requestID) {
|
|
return
|
|
}
|
|
handler.setUpstreamEnabled(writer, request, name, action == "enable", requestID, identity)
|
|
return
|
|
}
|
|
if name, _, ok := matchNamedAction(request.URL.Path, routingPrefix, "switch"); ok {
|
|
if !requireMethod(writer, request, http.MethodPost, requestID) {
|
|
return
|
|
}
|
|
handler.switchRouting(writer, request, name, requestID, identity)
|
|
return
|
|
}
|
|
|
|
writeTransportProblem(writer, http.StatusNotFound, "NOT_FOUND", "Not found", "the requested endpoint does not exist", requestID)
|
|
}
|
|
|
|
func (handler *Handler) getStatus(writer http.ResponseWriter, request *http.Request, requestID string) {
|
|
status, err := handler.service.Status(request.Context())
|
|
if err != nil {
|
|
writeServiceProblem(writer, err, requestID)
|
|
return
|
|
}
|
|
if status.Upstreams == nil {
|
|
status.Upstreams = []UpstreamStatus{}
|
|
}
|
|
if status.Workers == nil {
|
|
status.Workers = []WorkerStatus{}
|
|
}
|
|
writer.Header().Set(httpapi.HeaderRequestID, requestID)
|
|
_ = httpapi.WriteJSON(writer, http.StatusOK, status)
|
|
}
|
|
|
|
func (handler *Handler) readAudit(writer http.ResponseWriter, request *http.Request, requestID string) {
|
|
query, err := parseAuditQuery(request)
|
|
if err != nil {
|
|
writeTransportProblem(writer, http.StatusBadRequest, "INVALID_AUDIT_QUERY", "Invalid audit query", "audit pagination fields violate the API contract", requestID)
|
|
return
|
|
}
|
|
page, err := handler.service.ReadAudit(request.Context(), query)
|
|
if err != nil {
|
|
writeServiceProblem(writer, err, requestID)
|
|
return
|
|
}
|
|
if page.Records == nil {
|
|
page.Records = []AuditRecord{}
|
|
}
|
|
writer.Header().Set(httpapi.HeaderRequestID, requestID)
|
|
_ = httpapi.WriteJSON(writer, http.StatusOK, page)
|
|
}
|
|
|
|
func parseAuditQuery(request *http.Request) (AuditQuery, error) {
|
|
if request == nil || request.URL == nil {
|
|
return AuditQuery{}, ErrInvalidConfiguration
|
|
}
|
|
values := request.URL.Query()
|
|
query := AuditQuery{Limit: defaultAuditPageSize}
|
|
for name, value := range values {
|
|
if len(value) != 1 {
|
|
return AuditQuery{}, ErrInvalidConfiguration
|
|
}
|
|
switch name {
|
|
case "afterId":
|
|
parsed, err := strconv.ParseUint(value[0], 10, 64)
|
|
if err != nil {
|
|
return AuditQuery{}, ErrInvalidConfiguration
|
|
}
|
|
query.AfterID = parsed
|
|
case "limit":
|
|
parsed, err := strconv.ParseUint(value[0], 10, 32)
|
|
if err != nil || parsed == 0 || parsed > adminstate.MaxPageSize {
|
|
return AuditQuery{}, ErrInvalidConfiguration
|
|
}
|
|
query.Limit = int(parsed)
|
|
default:
|
|
return AuditQuery{}, ErrInvalidConfiguration
|
|
}
|
|
}
|
|
if err := query.Validate(); err != nil {
|
|
return AuditQuery{}, err
|
|
}
|
|
return query, nil
|
|
}
|
|
|
|
func (handler *Handler) setUpstreamEnabled(
|
|
writer http.ResponseWriter,
|
|
request *http.Request,
|
|
name string,
|
|
enabled bool,
|
|
requestID string,
|
|
identity httpsecurity.Identity,
|
|
) {
|
|
result, err := handler.service.SetUpstreamEnabled(request.Context(), SetUpstreamCommand{
|
|
RequestID: requestID,
|
|
ActorID: identity.ClientID,
|
|
SourceIP: identity.SourceIP,
|
|
Name: name,
|
|
Enabled: enabled,
|
|
})
|
|
if err != nil {
|
|
writeServiceProblem(writer, err, requestID)
|
|
return
|
|
}
|
|
writeMutation(writer, result, requestID)
|
|
}
|
|
|
|
func (handler *Handler) switchRouting(
|
|
writer http.ResponseWriter,
|
|
request *http.Request,
|
|
name string,
|
|
requestID string,
|
|
identity httpsecurity.Identity,
|
|
) {
|
|
var command SwitchCommand
|
|
if err := httpapi.DecodeJSON(writer, request, handler.maxBodyBytes, &command); err != nil {
|
|
writeDecodeProblem(writer, err, requestID)
|
|
return
|
|
}
|
|
if command.ExpectedCurrent == "" || command.Target == "" ||
|
|
len(command.ExpectedCurrent) > maxResourceNameBytes || len(command.Target) > maxResourceNameBytes ||
|
|
len(command.Reason) > 512 {
|
|
writeTransportProblem(writer, http.StatusUnprocessableEntity, "INVALID_SWITCH", "Invalid routing switch", "routing switch fields violate the API contract", requestID)
|
|
return
|
|
}
|
|
command.RequestID = requestID
|
|
command.ActorID = identity.ClientID
|
|
command.SourceIP = identity.SourceIP
|
|
command.Name = name
|
|
result, err := handler.service.SwitchRouting(request.Context(), command)
|
|
if err != nil {
|
|
writeServiceProblem(writer, err, requestID)
|
|
return
|
|
}
|
|
writeMutation(writer, result, requestID)
|
|
}
|
|
|
|
func (handler *Handler) reload(
|
|
writer http.ResponseWriter,
|
|
request *http.Request,
|
|
requestID string,
|
|
identity httpsecurity.Identity,
|
|
) {
|
|
result, err := handler.service.ReloadConfiguration(request.Context(), ReloadCommand{
|
|
RequestID: requestID,
|
|
ActorID: identity.ClientID,
|
|
SourceIP: identity.SourceIP,
|
|
})
|
|
if err != nil {
|
|
writeServiceProblem(writer, err, requestID)
|
|
return
|
|
}
|
|
writeMutation(writer, result, requestID)
|
|
}
|
|
|
|
func writeMutation(writer http.ResponseWriter, result MutationResult, requestID string) {
|
|
result.RequestID = requestID
|
|
writer.Header().Set(httpapi.HeaderRequestID, requestID)
|
|
_ = httpapi.WriteJSON(writer, http.StatusOK, result)
|
|
}
|
|
|
|
func matchNamedAction(path, prefix string, actions ...string) (string, string, bool) {
|
|
if !strings.HasPrefix(path, prefix) {
|
|
return "", "", false
|
|
}
|
|
remainder := strings.TrimPrefix(path, prefix)
|
|
name, action, ok := strings.Cut(remainder, "/")
|
|
if !ok || name == "" || len(name) > maxResourceNameBytes || strings.Contains(action, "/") {
|
|
return "", "", false
|
|
}
|
|
for _, allowed := range actions {
|
|
if action == allowed {
|
|
return name, action, true
|
|
}
|
|
}
|
|
return "", "", false
|
|
}
|
|
|
|
func requireMethod(writer http.ResponseWriter, request *http.Request, allowed, requestID string) bool {
|
|
if request.Method == allowed {
|
|
return true
|
|
}
|
|
writer.Header().Set("Allow", allowed)
|
|
writeTransportProblem(writer, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Method not allowed", fmt.Sprintf("use %s for this endpoint", allowed), requestID)
|
|
return false
|
|
}
|
|
|
|
func writeDecodeProblem(writer http.ResponseWriter, err error, requestID string) {
|
|
if errors.Is(err, httpapi.ErrUnsupportedMediaType) {
|
|
writeTransportProblem(writer, http.StatusUnsupportedMediaType, "UNSUPPORTED_MEDIA_TYPE", "Unsupported media type", "Content-Type must be application/json", requestID)
|
|
return
|
|
}
|
|
if errors.Is(err, httpapi.ErrBodyTooLarge) {
|
|
writeTransportProblem(writer, http.StatusRequestEntityTooLarge, "REQUEST_BODY_TOO_LARGE", "Request body too large", "request body exceeds the configured limit", requestID)
|
|
return
|
|
}
|
|
writeTransportProblem(writer, http.StatusBadRequest, "INVALID_JSON", "Invalid JSON", "request body must be one valid JSON document with no unknown fields", requestID)
|
|
}
|
|
|
|
func writeServiceProblem(writer http.ResponseWriter, err error, requestID string) {
|
|
switch {
|
|
case errors.Is(err, ErrNotFound):
|
|
writeTransportProblem(writer, http.StatusNotFound, "NOT_FOUND", "Not found", "the requested resource does not exist", requestID)
|
|
case errors.Is(err, ErrConflict):
|
|
writeTransportProblem(writer, http.StatusConflict, "CONFLICT", "Mutation conflict", "the authoritative state changed before the mutation committed", requestID)
|
|
case errors.Is(err, ErrInvalidConfiguration):
|
|
writeTransportProblem(writer, http.StatusUnprocessableEntity, "INVALID_CONFIGURATION", "Invalid configuration", "the new configuration did not pass validation", requestID)
|
|
case errors.Is(err, ErrUnavailable):
|
|
writeTransportProblem(writer, http.StatusServiceUnavailable, "SERVICE_UNAVAILABLE", "Service unavailable", "the authoritative service is temporarily unavailable", requestID)
|
|
default:
|
|
writeTransportProblem(writer, http.StatusInternalServerError, "INTERNAL_ERROR", "Internal server error", "the request could not be completed", requestID)
|
|
}
|
|
}
|
|
|
|
func writeTransportProblem(writer http.ResponseWriter, status int, code, title, detail, requestID string) {
|
|
httpapi.WriteProblem(writer, httpapi.NewProblem(status, code, title, detail, requestID))
|
|
}
|