package caddysnake
import (
"net/http"
"os"
"path/filepath"
"sync"
"time"
"github.com/fsnotify/fsnotify"
"go.uber.org/zap"
)
// watchDirRecursive adds all directories under root to the fsnotify watcher.
// It is used by both AutoreloadableApp and DynamicApp.
func watchDirRecursive(watcher *fsnotify.Watcher, root string, logger *zap.Logger) {
if err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
return nil
}
if addErr := watcher.Add(path); addErr != nil {
logger.Warn("autoreload: failed to watch directory",
zap.String("path", path),
zap.Error(addErr),
)
}
return nil
}); err != nil {
logger.Warn("autoreload: failed to walk working directory",
zap.String("path", root),
zap.Error(err),
)
}
}
// isPythonFileEvent returns true if the event is a write/create/remove/rename
// of a .py file.
func isPythonFileEvent(event fsnotify.Event) bool {
if filepath.Ext(event.Name) != ".py" {
return false
}
return event.Has(fsnotify.Write) || event.Has(fsnotify.Create) ||
event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename)
}
// handleNewDirEvent checks if the event is a newly created directory and adds
// it to the watcher if appropriate.
func handleNewDirEvent(event fsnotify.Event, watcher *fsnotify.Watcher) {
if !event.Has(fsnotify.Create) {
return
}
info, err := os.Stat(event.Name)
if err != nil || !info.IsDir() {
return
}
if err := watcher.Add(event.Name); err != nil {
return
}
}
// AutoreloadableApp wraps an AppServer to support hot-reloading when Python
// files in the working directory change. It watches for .py file modifications
// and reloads the app after a debounce period to group rapid changes.
type AutoreloadableApp struct {
mu sync.RWMutex
app AppServer
factory func() (AppServer, error)
watcher *fsnotify.Watcher
stopCh chan struct{}
logger *zap.Logger
workingDir string
exitOnReloadFailure func(code int) // if set, process exits on reload failure instead of serving 500
}
// NewAutoreloadableApp creates an AutoreloadableApp that wraps the given app and
// starts a filesystem watcher on the working directory. When any .py file changes,
// the app is reloaded after a 500ms debounce window.
// If exitOnReloadFailure is non-nil, it is called with code 1 when a reload fails
// (e.g. app deleted), so the process can terminate and stop serving requests.
func NewAutoreloadableApp(
app AppServer,
workingDir string,
factory func() (AppServer, error),
logger *zap.Logger,
exitOnReloadFailure func(code int),
) (*AutoreloadableApp, error) {
watcher, err := fsnotify.NewWatcher()
if err != nil {
return nil, err
}
a := &AutoreloadableApp{
app: app,
factory: factory,
watcher: watcher,
stopCh: make(chan struct{}),
logger: logger,
workingDir: workingDir,
exitOnReloadFailure: exitOnReloadFailure,
}
watchDirRecursive(watcher, workingDir, logger)
go a.watch()
logger.Info("autoreload enabled", zap.String("working_dir", workingDir))
return a, nil
}
// watch runs in a goroutine and listens for filesystem events.
// It debounces rapid changes (e.g. editor save + format) into a single reload.
func (a *AutoreloadableApp) watch() {
var debounceTimer *time.Timer
const debounceDuration = 500 * time.Millisecond
for {
select {
case event, ok := <-a.watcher.Events:
if !ok {
return
}
if !isPythonFileEvent(event) {
handleNewDirEvent(event, a.watcher)
continue
}
a.logger.Debug("python file changed",
zap.String("file", event.Name),
zap.String("op", event.Op.String()),
)
if debounceTimer != nil {
debounceTimer.Stop()
}
debounceTimer = time.AfterFunc(debounceDuration, func() {
a.reload()
})
case err, ok := <-a.watcher.Errors:
if !ok {
return
}
a.logger.Error("autoreload watcher error", zap.Error(err))
case <-a.stopCh:
if debounceTimer != nil {
debounceTimer.Stop()
}
return
}
}
}
// reload performs the actual app reload by stopping the old worker processes
// and starting new ones via the factory function.
func (a *AutoreloadableApp) reload() {
a.logger.Info("reloading python app due to file changes")
// Create new app OUTSIDE lock to avoid blocking requests
newApp, err := a.factory()
if err != nil {
a.logger.Error("failed to reload python app", zap.Error(err))
if a.exitOnReloadFailure != nil {
a.exitOnReloadFailure(1)
}
a.mu.Lock()
a.app = &errorApp{err: err}
a.mu.Unlock()
return
}
// Swap under lock (fast operation)
a.mu.Lock()
oldApp := a.app
a.app = newApp
a.mu.Unlock()
a.logger.Info("python app reloaded successfully")
// Cleanup old app OUTSIDE lock. The write lock above guarantees all
// in-flight requests using oldApp have completed before the swap.
if err := oldApp.Cleanup(); err != nil {
a.logger.Error("failed to cleanup old python app during reload", zap.Error(err))
}
}
// HandleRequest forwards the request to the underlying app while holding a read
// lock. This ensures the app isn't swapped mid-request.
func (a *AutoreloadableApp) HandleRequest(w http.ResponseWriter, r *http.Request) error {
a.mu.RLock()
defer a.mu.RUnlock()
return a.app.HandleRequest(w, r)
}
// Cleanup stops the filesystem watcher and cleans up the underlying app.
func (a *AutoreloadableApp) Cleanup() error {
close(a.stopCh)
a.watcher.Close()
a.mu.RLock()
app := a.app
a.mu.RUnlock()
return app.Cleanup()
}
// errorApp is a stub AppServer returned when a reload fails.
// It returns HTTP 503 for all requests until the next successful reload.
type errorApp struct {
err error
}
func (e *errorApp) HandleRequest(w http.ResponseWriter, r *http.Request) error {
w.WriteHeader(http.StatusServiceUnavailable)
_, _ = w.Write([]byte("Service temporarily unavailable"))
return nil
}
func (e *errorApp) Cleanup() error {
return nil
}
package caddysnake
import (
"bufio"
"errors"
"fmt"
"io"
"math"
"net"
"os"
"path/filepath"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"time"
)
// Env vars passed to Python workers for cache access.
const (
EnvCaddysnakeCacheAddr = "CADDYSNAKE_CACHE_ADDR"
EnvCaddysnakeWorkerInterface = "CADDYSNAKE_WORKER_INTERFACE"
EnvCaddysnakeWorkerID = "CADDYSNAKE_WORKER_ID"
EnvCaddysnakeCacheTimeoutSeconds = "CADDYSNAKE_CACHE_TIMEOUT"
)
// DefaultCacheClientTimeoutSec is a read/connect timeout hint for cache clients (seconds); client may ignore.
const DefaultCacheClientTimeoutSec = 30
// cacheAddrUnixScheme prefixes CADDYSNAKE_CACHE_ADDR when listening on a Unix domain socket.
const cacheAddrUnixScheme = "unix://"
// Resource limits (conservative defaults).
const (
maxCacheKeyLen = 8192
maxCacheScalarLen = 1 << 20 // 1 MiB
maxCacheListElemLen = 1 << 20
maxCacheListLen = 100_000
maxCacheKeys = 1_000_000
maxRESPProtoLineBytes = 1 << 20
defaultCSKEYSLimit = 1000
maxCSKEYSLimit = 1000
maxSubscribeTimeoutSec = 300.0
)
var (
errCacheLimit = errors.New("limit")
errWrongType = errors.New("wrong type")
)
type entryKind int
const (
entryScalar entryKind = iota
entryList
entrySet
)
type cacheEntry struct {
kind entryKind
scalar []byte
list [][]byte
members map[string][]byte // set: dedup by string(member bytes)
expiry *time.Time // wall clock; nil = no expiry
}
func (e *cacheEntry) expired(now time.Time) bool {
if e == nil || e.expiry == nil {
return false
}
return !now.Before(*e.expiry)
}
type cacheStore struct {
mu sync.Mutex
data map[string]*cacheEntry
conds map[string]*sync.Cond // lazily created; all use &mu
closing bool
keyCap int //nolint:unused // reserved for future max-keys enforcement
}
func newCacheStore() *cacheStore {
s := &cacheStore{
data: make(map[string]*cacheEntry),
conds: make(map[string]*sync.Cond),
}
return s
}
func (s *cacheStore) condForKey(key string) *sync.Cond {
if c, ok := s.conds[key]; ok {
return c
}
c := sync.NewCond(&s.mu)
s.conds[key] = c
return c
}
func (s *cacheStore) dropCond(key string) {
delete(s.conds, key)
}
func (s *cacheStore) broadcastKey(key string) {
if c := s.conds[key]; c != nil {
c.Broadcast()
}
}
func (s *cacheStore) deleteEntryLocked(key string) {
delete(s.data, key)
s.dropCond(key)
}
func (s *cacheStore) shutdownLocked() {
s.closing = true
for _, c := range s.conds {
c.Broadcast()
}
}
func (s *cacheStore) Shutdown() {
s.mu.Lock()
s.shutdownLocked()
s.mu.Unlock()
}
func (s *cacheStore) entryLocked(k string, now time.Time) (*cacheEntry, bool) {
e := s.data[k]
if e == nil {
return nil, false
}
if e.expired(now) {
s.deleteEntryLocked(k)
return nil, false
}
return e, true
}
func (s *cacheStore) checkKeyLen(key []byte) error {
if len(key) == 0 || len(key) > maxCacheKeyLen {
return fmt.Errorf("%w: key size", errCacheLimit)
}
return nil
}
func sanitizeErrMsg(msg string) string {
msg = strings.ReplaceAll(msg, "\r", " ")
msg = strings.ReplaceAll(msg, "\n", " ")
return msg
}
func (s *cacheStore) checkScalarVal(v []byte) error {
if len(v) > maxCacheScalarLen {
return fmt.Errorf("%w: value size", errCacheLimit)
}
return nil
}
func (s *cacheStore) checkElem(v []byte) error {
if len(v) > maxCacheListElemLen {
return fmt.Errorf("%w: list element size", errCacheLimit)
}
return nil
}
func (s *cacheStore) enforceKeysCap() error {
if len(s.data) >= maxCacheKeys {
return fmt.Errorf("%w: too many keys", errCacheLimit)
}
return nil
}
func wallExpiry(ttlSec int64) *time.Time {
if ttlSec <= 0 {
return nil
}
t := time.Now().Add(time.Duration(ttlSec) * time.Second)
return &t
}
// Set overwrites any prior value (scalar or list).
func (s *cacheStore) Set(key, value []byte, ttlSec int64) error {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return err
}
if err := s.checkScalarVal(value); err != nil {
return err
}
k := string(key)
if e := s.data[k]; e != nil && e.expired(now) {
s.deleteEntryLocked(k)
e = nil
}
if _, ok := s.data[k]; !ok {
if err := s.enforceKeysCap(); err != nil {
return err
}
}
exp := wallExpiry(ttlSec)
s.data[k] = &cacheEntry{kind: entryScalar, scalar: append([]byte(nil), value...), expiry: exp}
s.broadcastKey(k) // wake pops if type changed
return nil
}
// Get returns value data and kind. ok=false on miss or invalid key.
func (s *cacheStore) Get(key []byte) (scalar []byte, list [][]byte, setMembers [][]byte, kind entryKind, ok bool) {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return nil, nil, nil, entryScalar, false
}
k := string(key)
e, ok := s.entryLocked(k, now)
if !ok {
return nil, nil, nil, entryScalar, false
}
switch e.kind {
case entryScalar:
return append([]byte(nil), e.scalar...), nil, nil, entryScalar, true
case entryList:
out := make([][]byte, len(e.list))
for i, b := range e.list {
out[i] = append([]byte(nil), b...)
}
return nil, out, nil, entryList, true
case entrySet:
out := setMembersSorted(e.members)
return nil, nil, out, entrySet, true
default:
return nil, nil, nil, entryScalar, false
}
}
func (s *cacheStore) Delete(key []byte) int {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return 0
}
k := string(key)
e := s.data[k]
if e == nil {
return 0
}
if e.expired(now) {
s.deleteEntryLocked(k)
return 0
}
s.broadcastKey(k)
s.deleteEntryLocked(k)
return 1
}
func (s *cacheStore) Append(key, value []byte) error {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return err
}
if err := s.checkElem(value); err != nil {
return err
}
k := string(key)
e := s.data[k]
if e != nil && e.expired(now) {
s.deleteEntryLocked(k)
e = nil
}
if e == nil {
if err := s.enforceKeysCap(); err != nil {
return err
}
if 1 > maxCacheListLen {
return fmt.Errorf("%w: list length", errCacheLimit)
}
s.data[k] = &cacheEntry{
kind: entryList,
list: [][]byte{append([]byte(nil), value...)},
expiry: nil,
}
s.broadcastKey(k)
return nil
}
// clear TTL on append (per spec)
noTTL := (*time.Time)(nil)
if e.kind == entrySet {
return errWrongType
}
if e.kind == entryScalar {
if 2 > maxCacheListLen {
return fmt.Errorf("%w: list length", errCacheLimit)
}
old := append([]byte(nil), e.scalar...)
s.data[k] = &cacheEntry{
kind: entryList,
list: [][]byte{old, append([]byte(nil), value...)},
expiry: noTTL,
}
s.broadcastKey(k)
return nil
}
if len(e.list) >= maxCacheListLen {
return fmt.Errorf("%w: list length", errCacheLimit)
}
e.list = append(e.list, append([]byte(nil), value...))
e.expiry = noTTL
s.broadcastKey(k)
return nil
}
// Pop returns (value, true) | (nil, false) for immediate nil (miss, scalar, timeout, cancelled).
func (s *cacheStore) Pop(key []byte, deadline *time.Time) ([]byte, bool) {
s.mu.Lock()
if err := s.checkKeyLen(key); err != nil {
s.mu.Unlock()
return nil, false
}
k := string(key)
timedOut := false
var timer *time.Timer
if deadline != nil {
d := time.Until(*deadline)
if d <= 0 {
s.mu.Unlock()
return nil, false
}
timer = time.AfterFunc(d, func() {
s.mu.Lock()
timedOut = true
s.broadcastKey(k)
s.mu.Unlock()
})
}
for {
if s.closing {
if timer != nil {
timer.Stop()
}
s.mu.Unlock()
return nil, false
}
now := time.Now()
e, exists := s.entryLocked(k, now)
if !exists {
if timer != nil {
timer.Stop()
}
s.mu.Unlock()
return nil, false
}
if e.kind == entryScalar || e.kind == entrySet {
if timer != nil {
timer.Stop()
}
s.mu.Unlock()
return nil, false
}
if len(e.list) > 0 {
v := e.list[0]
e.list = e.list[1:]
out := append([]byte(nil), v...)
if timer != nil {
timer.Stop()
}
s.mu.Unlock()
return out, true
}
if timedOut {
if timer != nil {
timer.Stop()
}
s.mu.Unlock()
return nil, false
}
cond := s.condForKey(k)
cond.Wait()
}
}
func setMembersSorted(m map[string][]byte) [][]byte {
if len(m) == 0 {
return nil
}
keys := make([]string, 0, len(m))
for k := range m {
keys = append(keys, k)
}
sortStrings(keys)
out := make([][]byte, len(keys))
for i, k := range keys {
out[i] = append([]byte(nil), m[k]...)
}
return out
}
func sortStrings(ss []string) {
sort.Strings(ss)
}
// SAdd returns 1 if member was added, 0 if already present.
func (s *cacheStore) SAdd(key, member []byte) (int, error) {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return 0, err
}
if err := s.checkElem(member); err != nil {
return 0, err
}
k := string(key)
mk := string(member)
e := s.data[k]
if e != nil && e.expired(now) {
s.deleteEntryLocked(k)
e = nil
}
if e != nil && e.kind != entrySet {
return 0, errWrongType
}
if e == nil {
if err := s.enforceKeysCap(); err != nil {
return 0, err
}
s.data[k] = &cacheEntry{
kind: entrySet,
members: map[string][]byte{mk: append([]byte(nil), member...)},
expiry: nil,
}
return 1, nil
}
if _, ok := e.members[mk]; ok {
return 0, nil
}
if len(e.members) >= maxCacheListLen {
return 0, fmt.Errorf("%w: set size", errCacheLimit)
}
e.members[mk] = append([]byte(nil), member...)
return 1, nil
}
// SRem returns 1 if member was removed, 0 if absent.
func (s *cacheStore) SRem(key, member []byte) (int, error) {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return 0, err
}
k := string(key)
e := s.data[k]
if e == nil || e.expired(now) {
if e != nil {
s.deleteEntryLocked(k)
}
return 0, nil
}
if e.kind != entrySet {
return 0, errWrongType
}
mk := string(member)
if _, ok := e.members[mk]; !ok {
return 0, nil
}
delete(e.members, mk)
if len(e.members) == 0 {
s.deleteEntryLocked(k)
}
return 1, nil
}
// SMembers returns sorted member copies; empty slice if key missing (Redis-aligned).
func (s *cacheStore) SMembers(key []byte) ([][]byte, error) {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return nil, err
}
k := string(key)
e, ok := s.entryLocked(k, now)
if !ok {
return [][]byte{}, nil
}
if e.kind != entrySet {
return nil, errWrongType
}
return setMembersSorted(e.members), nil
}
// SetNX returns 1 if key was set, 0 if key already exists (any non-expired kind).
func (s *cacheStore) SetNX(key, value []byte, ttlSec int64) (int, error) {
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if err := s.checkKeyLen(key); err != nil {
return 0, err
}
if err := s.checkScalarVal(value); err != nil {
return 0, err
}
k := string(key)
if e, ok := s.entryLocked(k, now); ok && e != nil {
return 0, nil
}
if err := s.enforceKeysCap(); err != nil {
return 0, err
}
exp := wallExpiry(ttlSec)
s.data[k] = &cacheEntry{kind: entryScalar, scalar: append([]byte(nil), value...), expiry: exp}
s.broadcastKey(k)
return 1, nil
}
// Keys returns up to limit key names matching prefix (sorted). Prefix must be non-empty.
func (s *cacheStore) Keys(prefix []byte, limit int) ([][]byte, error) {
if len(prefix) == 0 {
return nil, fmt.Errorf("%w: keys prefix required", errCacheLimit)
}
if limit <= 0 {
limit = defaultCSKEYSLimit
}
if limit > maxCSKEYSLimit {
limit = maxCSKEYSLimit
}
pfx := string(prefix)
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
var matched []string
for k, e := range s.data {
if e.expired(now) {
s.deleteEntryLocked(k)
continue
}
if !strings.HasPrefix(k, pfx) {
continue
}
matched = append(matched, k)
}
sortStrings(matched)
if len(matched) > limit {
matched = matched[:limit]
}
out := make([][]byte, len(matched))
for i, k := range matched {
out[i] = []byte(k)
}
return out, nil
}
// --- pub/sub (blocking one-shot receive) ---
type pubSubWaiter struct {
msg []byte
ready bool
cancelled bool
}
type pubSubChannel struct {
waiters []*pubSubWaiter
cond *sync.Cond
}
type pubSub struct {
mu sync.Mutex
channels map[string]*pubSubChannel
closing bool
}
func newPubSub() *pubSub {
ps := &pubSub{channels: make(map[string]*pubSubChannel)}
return ps
}
func (ps *pubSub) channelLocked(name string) *pubSubChannel {
ch, ok := ps.channels[name]
if !ok {
ch = &pubSubChannel{cond: sync.NewCond(&ps.mu)}
ps.channels[name] = ch
}
return ch
}
func (ps *pubSub) removeWaiter(ch *pubSubChannel, w *pubSubWaiter) {
for i, x := range ch.waiters {
if x == w {
ch.waiters = append(ch.waiters[:i], ch.waiters[i+1:]...)
return
}
}
}
func (ps *pubSub) Subscribe(channel []byte, deadline time.Time) ([]byte, bool) {
if err := checkPubSubName(channel); err != nil {
return nil, false
}
name := string(channel)
ps.mu.Lock()
if ps.closing {
ps.mu.Unlock()
return nil, false
}
ch := ps.channelLocked(name)
w := &pubSubWaiter{}
ch.waiters = append(ch.waiters, w)
timedOut := false
timer := time.AfterFunc(time.Until(deadline), func() {
ps.mu.Lock()
timedOut = true
ch.cond.Broadcast()
ps.mu.Unlock()
})
for !w.ready && !w.cancelled && !ps.closing && !timedOut {
ch.cond.Wait()
}
timer.Stop()
var out []byte
if w.ready {
out = append([]byte(nil), w.msg...)
}
ps.removeWaiter(ch, w)
if len(ch.waiters) == 0 {
delete(ps.channels, name)
}
ps.mu.Unlock()
if w.ready {
return out, true
}
return nil, false
}
func (ps *pubSub) Publish(channel, message []byte) (int, error) {
if err := checkPubSubName(channel); err != nil {
return 0, err
}
if len(message) > maxCacheScalarLen {
return 0, fmt.Errorf("%w: message size", errCacheLimit)
}
name := string(channel)
ps.mu.Lock()
defer ps.mu.Unlock()
if ps.closing {
return 0, nil
}
ch, ok := ps.channels[name]
if !ok || len(ch.waiters) == 0 {
return 0, nil
}
n := 0
for _, w := range ch.waiters {
if w.ready {
continue
}
w.msg = append([]byte(nil), message...)
w.ready = true
n++
}
ch.cond.Broadcast()
return n, nil
}
func (ps *pubSub) Shutdown() {
ps.mu.Lock()
ps.closing = true
for _, ch := range ps.channels {
for _, w := range ch.waiters {
w.cancelled = true
}
ch.cond.Broadcast()
}
ps.mu.Unlock()
}
func checkPubSubName(channel []byte) error {
if len(channel) == 0 || len(channel) > maxCacheKeyLen {
return fmt.Errorf("%w: channel size", errCacheLimit)
}
return nil
}
// --- RESP ---
func respWriteSimpleString(w *bufio.Writer, s string) error {
if _, err := fmt.Fprintf(w, "+%s\r\n", s); err != nil {
return err
}
return w.Flush()
}
func respWriteError(w *bufio.Writer, msg string) error {
if _, err := fmt.Fprintf(w, "-ERR %s\r\n", sanitizeErrMsg(msg)); err != nil {
return err
}
return w.Flush()
}
func respWriteInt(w *bufio.Writer, n int64) error {
if _, err := fmt.Fprintf(w, ":%d\r\n", n); err != nil {
return err
}
return w.Flush()
}
func respWriteBulk(w *bufio.Writer, b []byte) error {
if b == nil {
if _, err := io.WriteString(w, "$-1\r\n"); err != nil {
return err
}
return w.Flush()
}
if _, err := fmt.Fprintf(w, "$%d\r\n", len(b)); err != nil {
return err
}
if _, err := w.Write(b); err != nil {
return err
}
if _, err := io.WriteString(w, "\r\n"); err != nil {
return err
}
return w.Flush()
}
func respWriteBulkString(w *bufio.Writer, s string) error {
return respWriteBulk(w, []byte(s))
}
func respWriteArrayHeader(w *bufio.Writer, n int) error {
if _, err := fmt.Fprintf(w, "*%d\r\n", n); err != nil {
return err
}
return nil
}
// respWriteArrayOfBulks writes *n followed by n bulk strings (no trailing flush until end).
func respWriteArrayOfBulks(w *bufio.Writer, elems [][]byte) error {
if err := respWriteArrayHeader(w, len(elems)); err != nil {
return err
}
for _, b := range elems {
if _, err := fmt.Fprintf(w, "$%d\r\n", len(b)); err != nil {
return err
}
if _, err := w.Write(b); err != nil {
return err
}
if _, err := io.WriteString(w, "\r\n"); err != nil {
return err
}
}
return w.Flush()
}
func respReadLine(r *bufio.Reader) ([]byte, error) {
line, err := r.ReadBytes('\n')
if err != nil {
return nil, err
}
if len(line) < 2 || line[len(line)-2] != '\r' {
return nil, fmt.Errorf("invalid line ending")
}
if len(line)-2 > maxRESPProtoLineBytes {
return nil, errCacheLimit
}
return line[:len(line)-2], nil
}
func respReadBulk(r *bufio.Reader) ([]byte, error) {
line, err := respReadLine(r)
if err != nil {
return nil, err
}
if len(line) < 1 || line[0] != '$' {
return nil, fmt.Errorf("expected bulk")
}
n, err := strconv.Atoi(string(line[1:]))
if err != nil {
return nil, err
}
if n == -1 {
return nil, nil // null bulk in request? treat as empty
}
if n < 0 || n > maxCacheScalarLen {
return nil, errCacheLimit
}
buf := make([]byte, n+2)
if _, err := io.ReadFull(r, buf); err != nil {
return nil, err
}
if buf[n] != '\r' || buf[n+1] != '\n' {
return nil, fmt.Errorf("bulk trailer")
}
return buf[:n], nil
}
func respReadArray(r *bufio.Reader) ([][]byte, error) {
line, err := respReadLine(r)
if err != nil {
return nil, err
}
if len(line) < 1 || line[0] != '*' {
return nil, fmt.Errorf("expected array")
}
n, err := strconv.Atoi(string(line[1:]))
if err != nil || n < 1 || n > 32 {
return nil, fmt.Errorf("bad array len")
}
out := make([][]byte, n)
for i := 0; i < n; i++ {
b, err := respReadBulk(r)
if err != nil {
return nil, err
}
out[i] = b
}
return out, nil
}
// --- IPC server (Unix socket on unix-like OS; loopback TCP on Windows) ---
type cacheServer struct {
store *cacheStore
pubsub *pubSub
ln net.Listener
done chan struct{}
closeMu sync.Mutex
closed bool
// addr is CADDYSNAKE_CACHE_ADDR: "unix://" + filesystem path, or "127.0.0.1:<port>" on Windows.
addr string
sockDir string // non-empty when using a Unix socket — removed on Close
}
func startCacheServer() (*cacheServer, error) {
if runtime.GOOS == "windows" {
return startCacheServerTCPOnly()
}
return startCacheServerUnixSocket()
}
func startCacheServerTCPOnly() (*cacheServer, error) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return nil, err
}
s := &cacheServer{
store: newCacheStore(),
pubsub: newPubSub(),
ln: ln,
done: make(chan struct{}),
addr: ln.Addr().String(),
}
go s.acceptLoop()
return s, nil
}
func startCacheServerUnixSocket() (*cacheServer, error) {
dir, err := os.MkdirTemp("", "cs-*")
if err != nil {
return nil, err
}
if chErr := os.Chmod(dir, 0o700); chErr != nil {
os.RemoveAll(dir)
return nil, chErr
}
path := filepath.Join(dir, "cache.sock")
_ = os.Remove(path)
ln, err := net.Listen("unix", path)
if err != nil {
os.RemoveAll(dir)
return nil, err
}
absPath, err := filepath.Abs(path)
if err != nil {
ln.Close()
os.RemoveAll(dir)
return nil, err
}
envAddr := cacheAddrUnixScheme + filepath.ToSlash(absPath)
s := &cacheServer{
store: newCacheStore(),
pubsub: newPubSub(),
ln: ln,
done: make(chan struct{}),
addr: envAddr,
sockDir: dir,
}
go s.acceptLoop()
return s, nil
}
func (s *cacheServer) Addr() string { return s.addr }
func (s *cacheServer) Close() error {
s.closeMu.Lock()
if s.closed {
s.closeMu.Unlock()
return nil
}
s.closed = true
s.closeMu.Unlock()
s.store.Shutdown()
s.pubsub.Shutdown()
err := s.ln.Close()
<-s.done
if s.sockDir != "" {
_ = os.RemoveAll(s.sockDir)
}
return err
}
func (s *cacheServer) acceptLoop() {
defer close(s.done)
for {
conn, err := s.ln.Accept()
if err != nil {
return
}
if tcp, ok := conn.(*net.TCPConn); ok {
_ = tcp.SetNoDelay(true)
}
go s.handleConn(conn)
}
}
func (s *cacheServer) handleConn(conn net.Conn) {
defer conn.Close()
r := bufio.NewReader(conn)
w := bufio.NewWriter(conn)
for {
parts, err := respReadArray(r)
if err != nil {
if errors.Is(err, io.EOF) {
return
}
_ = respWriteError(w, err.Error())
return
}
if len(parts) == 0 {
_ = respWriteError(w, "empty command")
return
}
cmd := strings.ToUpper(string(parts[0]))
switch cmd {
case "CSQUIT":
_ = respWriteSimpleString(w, "OK")
return
case "CSGET":
if len(parts) != 2 {
_ = respWriteError(w, "wrong number of arguments for CSGET")
return
}
scalar, list, _, kind, ok := s.store.Get(parts[1])
if !ok {
_ = respWriteBulk(w, nil) // $-1
continue
}
if kind == entrySet {
_ = respWriteError(w, errWrongType.Error())
continue
}
if kind == entryScalar {
_ = respWriteBulk(w, scalar)
continue
}
if len(list) == 0 {
if err := respWriteArrayHeader(w, 0); err != nil {
return
}
_ = w.Flush()
continue
}
_ = respWriteArrayOfBulks(w, list)
case "CSDEL":
if len(parts) != 2 {
_ = respWriteError(w, "wrong number of arguments for CSDEL")
return
}
n := s.store.Delete(parts[1])
_ = respWriteInt(w, int64(n))
case "CSSET":
if len(parts) != 3 && len(parts) != 4 {
_ = respWriteError(w, "wrong number of arguments for CSSET")
return
}
var ttl int64
if len(parts) == 4 && len(parts[3]) > 0 {
t, err := strconv.ParseInt(string(parts[3]), 10, 64)
if err != nil || t < 0 {
_ = respWriteError(w, "invalid TTL")
return
}
ttl = t
}
if err := s.store.Set(parts[1], parts[2], ttl); err != nil {
_ = respWriteError(w, err.Error())
continue
}
_ = respWriteSimpleString(w, "OK")
case "CSAPPEND":
if len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSAPPEND")
return
}
if err := s.store.Append(parts[1], parts[2]); err != nil {
if errors.Is(err, errWrongType) {
_ = respWriteError(w, errWrongType.Error())
} else {
_ = respWriteError(w, err.Error())
}
continue
}
_ = respWriteSimpleString(w, "OK")
case "CSPOP":
if len(parts) != 2 && len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSPOP")
return
}
var dl *time.Time
if len(parts) == 3 && len(parts[2]) > 0 {
sec, err := strconv.ParseFloat(string(parts[2]), 64)
if err != nil || sec < 0 {
_ = respWriteError(w, "invalid timeout")
return
}
t := time.Now().Add(time.Duration(sec * float64(time.Second)))
dl = &t
}
v, ok := s.store.Pop(parts[1], dl)
if !ok {
_ = respWriteBulk(w, nil)
continue
}
_ = respWriteBulk(w, v)
case "CSSADD":
if len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSSADD")
return
}
n, err := s.store.SAdd(parts[1], parts[2])
if err != nil {
if errors.Is(err, errWrongType) {
_ = respWriteError(w, errWrongType.Error())
} else {
_ = respWriteError(w, err.Error())
}
continue
}
_ = respWriteInt(w, int64(n))
case "CSSREM":
if len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSSREM")
return
}
n, err := s.store.SRem(parts[1], parts[2])
if err != nil {
if errors.Is(err, errWrongType) {
_ = respWriteError(w, errWrongType.Error())
} else {
_ = respWriteError(w, err.Error())
}
continue
}
_ = respWriteInt(w, int64(n))
case "CSSMEMBERS":
if len(parts) != 2 {
_ = respWriteError(w, "wrong number of arguments for CSSMEMBERS")
return
}
members, err := s.store.SMembers(parts[1])
if err != nil {
if errors.Is(err, errWrongType) {
_ = respWriteError(w, errWrongType.Error())
} else {
_ = respWriteError(w, err.Error())
}
continue
}
if len(members) == 0 {
if err := respWriteArrayHeader(w, 0); err != nil {
return
}
_ = w.Flush()
continue
}
_ = respWriteArrayOfBulks(w, members)
case "CSSETNX":
if len(parts) != 3 && len(parts) != 4 {
_ = respWriteError(w, "wrong number of arguments for CSSETNX")
return
}
var ttl int64
if len(parts) == 4 && len(parts[3]) > 0 {
t, err := strconv.ParseInt(string(parts[3]), 10, 64)
if err != nil || t < 0 {
_ = respWriteError(w, "invalid TTL")
return
}
ttl = t
}
n, err := s.store.SetNX(parts[1], parts[2], ttl)
if err != nil {
_ = respWriteError(w, err.Error())
continue
}
_ = respWriteInt(w, int64(n))
case "CSKEYS":
if len(parts) != 1 && len(parts) != 2 && len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSKEYS")
return
}
prefix := []byte{}
limit := defaultCSKEYSLimit
if len(parts) >= 2 {
prefix = parts[1]
}
if len(parts) == 3 {
l, err := strconv.Atoi(string(parts[2]))
if err != nil || l < 0 {
_ = respWriteError(w, "invalid limit")
return
}
limit = l
}
keys, err := s.store.Keys(prefix, limit)
if err != nil {
_ = respWriteError(w, err.Error())
continue
}
if len(keys) == 0 {
if err := respWriteArrayHeader(w, 0); err != nil {
return
}
_ = w.Flush()
continue
}
_ = respWriteArrayOfBulks(w, keys)
case "CSPUBLISH":
if len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSPUBLISH")
return
}
n, err := s.pubsub.Publish(parts[1], parts[2])
if err != nil {
_ = respWriteError(w, err.Error())
continue
}
_ = respWriteInt(w, int64(n))
case "CSSUBSCRIBE":
if len(parts) != 3 {
_ = respWriteError(w, "wrong number of arguments for CSSUBSCRIBE")
return
}
sec, err := strconv.ParseFloat(string(parts[2]), 64)
if err != nil || math.IsNaN(sec) || math.IsInf(sec, 0) || sec <= 0 || sec > maxSubscribeTimeoutSec {
_ = respWriteError(w, "invalid timeout")
return
}
deadline := time.Now().Add(time.Duration(sec * float64(time.Second)))
v, ok := s.pubsub.Subscribe(parts[1], deadline)
if !ok {
_ = respWriteBulk(w, nil)
continue
}
_ = respWriteBulk(w, v)
default:
_ = respWriteError(w, fmt.Sprintf("unknown command %q", cmd))
return
}
}
}
// Caddy plugin to serve Python apps.
package caddysnake
import (
"context"
_ "embed"
"encoding/json"
"errors"
"fmt"
"log"
"net"
"net/http"
"net/http/httputil"
"os"
"os/exec"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
"github.com/caddyserver/caddy/v2"
"github.com/caddyserver/caddy/v2/caddyconfig"
"github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
"github.com/caddyserver/caddy/v2/caddyconfig/httpcaddyfile"
caddycmd "github.com/caddyserver/caddy/v2/cmd"
"github.com/caddyserver/caddy/v2/modules/caddyhttp"
"github.com/caddyserver/certmagic"
"github.com/spf13/cobra"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"github.com/caddyserver/caddy/v2/modules/caddyhttp/encode"
_ "github.com/caddyserver/caddy/v2/modules/caddyhttp/encode/gzip"
_ "github.com/caddyserver/caddy/v2/modules/caddyhttp/encode/zstd"
_ "github.com/caddyserver/caddy/v2/modules/caddyhttp/fileserver"
)
//go:embed caddysnake.py
var caddysnake_py string
// Hop-internal headers: set only by the Go module after stripping client-supplied
// values. The Python worker uses these for REMOTE_ADDR / ASGI client.
const (
caddySnakeRemoteAddrHeader = "Caddy-Snake-Remote-Addr"
caddySnakeRemotePortHeader = "Caddy-Snake-Remote-Port"
)
// setPythonWorkerOutboundHeaders configures the outbound request to the Python
// worker: target URL, standard X-Forwarded-* (preserve inbound X-Forwarded-For
// chain, then append this hop per [httputil.ProxyRequest.SetXForwarded]), and
// trusted Caddy-Snake-Remote-* from the inbound RemoteAddr.
func setPythonWorkerOutboundHeaders(pr *httputil.ProxyRequest, dialAddr string) {
pr.Out.URL.Scheme = "http"
pr.Out.URL.Host = dialAddr
pr.Out.Header["X-Forwarded-For"] = pr.In.Header["X-Forwarded-For"]
pr.SetXForwarded()
pr.Out.Header.Del(caddySnakeRemoteAddrHeader)
pr.Out.Header.Del(caddySnakeRemotePortHeader)
host, port, err := net.SplitHostPort(pr.In.RemoteAddr)
if err != nil {
host = pr.In.RemoteAddr
port = "0"
}
pr.Out.Header.Set(caddySnakeRemoteAddrHeader, host)
pr.Out.Header.Set(caddySnakeRemotePortHeader, port)
}
// AppServer defines the interface to interacting with a WSGI or ASGI server
type AppServer interface {
Cleanup() error
HandleRequest(w http.ResponseWriter, r *http.Request) error
}
// DefaultStartTimeout is how long Provision waits for each Python worker to
// become ready when start_timeout is omitted.
const DefaultStartTimeout = 120 * time.Second
// startTimeoutWarnAfter is when a slow-start warning is logged if the configured
// start_timeout allows waiting longer. Overridable via
// CADDYSNAKE_START_TIMEOUT_WARN_AFTER (duration string) for integration tests.
var startTimeoutWarnAfter = DefaultStartTimeout
// errWorkerExited indicates the Python worker process exited before its socket
// or port file became ready.
var errWorkerExited = errors.New("python worker exited before becoming ready")
func applyStartTimeoutWarnAfterEnv() {
if v := os.Getenv("CADDYSNAKE_START_TIMEOUT_WARN_AFTER"); v != "" {
d, err := time.ParseDuration(v)
if err == nil && d > 0 {
startTimeoutWarnAfter = d
}
}
}
// parseCLIEnvVars parses repeated --env-var NAME=VALUE flags into a map.
func parseCLIEnvVars(flags []string) (map[string]string, error) {
if len(flags) == 0 {
return nil, nil
}
out := make(map[string]string, len(flags))
for _, raw := range flags {
name, value, ok := strings.Cut(raw, "=")
if !ok || name == "" {
return nil, fmt.Errorf("invalid --env-var %q (want NAME=VALUE)", raw)
}
if err := validateEnvVarName(name); err != nil {
return nil, fmt.Errorf("invalid --env-var name %q: %w", name, err)
}
out[name] = value
}
return out, nil
}
// CaddySnake is an HTTP handler that serves Python WSGI, ASGI, or ESGI apps
// by spawning worker subprocesses and proxying requests to them.
//
// Exactly one of module_wsgi, module_asgi, or module_esgi must be set.
// Caddyfile: the `python` directive (see https://caddy-snake.readthedocs.io/).
type CaddySnake struct {
// WSGI app import path as "module:variable" (e.g. "main:app" or "mysite.wsgi:application").
// Mutually exclusive with module_asgi and module_esgi. Supports Caddy placeholders for dynamic loading.
ModuleWsgi string `json:"module_wsgi,omitempty"`
// ASGI app import path as "module:variable" (e.g. "main:app" for FastAPI/Starlette).
// Mutually exclusive with module_wsgi and module_esgi. Supports Caddy placeholders for dynamic loading.
ModuleAsgi string `json:"module_asgi,omitempty"`
// ESGI app import path as "module:variable" (synchronous application(scope, protocol) callable).
// Mutually exclusive with module_wsgi and module_asgi. Supports Caddy placeholders for dynamic loading.
ModuleEsgi string `json:"module_esgi,omitempty"`
// Python worker runtime at the gateway boundary.
// WSGI: "sync" (default) or "gevent". ESGI: "gevent" only (default).
// ASGI: "native" or "uvloop" (default when omitted: "uvloop").
Runtime string `json:"runtime,omitempty"`
// ASGI lifespan protocol: "on" or "off" (default). Only applies with module_asgi.
Lifespan string `json:"lifespan,omitempty"`
// Working directory for the Python app (imports, relative paths, and autoreload watch root).
// Supports Caddy placeholders for dynamic resolution per request.
WorkingDir string `json:"working_dir,omitempty"`
// Path to a Python virtual environment; its site-packages are added to sys.path.
// Caddyfile subdirective name is "venv". Supports Caddy placeholders.
VenvPath string `json:"venv_path,omitempty"`
// Number of worker processes to spawn. Defaults to GOMAXPROCS (CPU count) when empty or "0".
Workers string `json:"workers,omitempty"`
// How long Provision waits for each worker to become ready.
// Empty uses 120s; "-1", "forever", "none", "inf", or "indefinite" wait indefinitely.
// Caddyfile: start_timeout. CLI: --start-timeout (use --start-timeout=-1 or forever).
StartTimeout string `json:"start_timeout,omitempty"`
// When set to "on", watch .py files under working_dir and reload workers on changes.
// Caddyfile: the bare "autoreload" subdirective.
Autoreload string `json:"autoreload,omitempty"`
// Path to the Python interpreter. Defaults to venv/bin/python when venv_path is set, else system python3.
PythonPath string `json:"python_path,omitempty"`
// Dotenv-style files loaded into each worker environment (later files override earlier keys).
// Relative paths resolve against working_dir when set. Caddyfile: repeatable "env_file" subdirective.
EnvFiles []string `json:"env_files,omitempty"`
// Individual environment variables for workers. Applied after env_files (overrides file values).
// Caddyfile: repeatable "env_var NAME value". Cannot set PYTHONUNBUFFERED or CADDYSNAKE_*.
EnvVars map[string]string `json:"env_vars,omitempty"`
logger *zap.Logger
app AppServer
cacheSrv *cacheServer
}
// parseStartTimeout parses a Caddyfile/JSON/CLI start_timeout value.
// Empty means DefaultStartTimeout. Indefinite wait (until the worker is ready
// or the process exits) is "-1", or the aliases "forever", "none", "inf", and
// "indefinite". Other values use Caddy durations.
//
// CLI note: pass indefinite as --start-timeout=-1 (equals form) or
// --start-timeout forever. A bare "--start-timeout -1" is parsed as flags by
// Cobra/pflag, not as a string value.
func parseStartTimeout(s string) (time.Duration, error) {
if s == "" {
return DefaultStartTimeout, nil
}
switch strings.ToLower(s) {
case "-1", "forever", "none", "inf", "indefinite":
return -1, nil
}
d, err := caddy.ParseDuration(s)
if err != nil {
return 0, fmt.Errorf("invalid start_timeout: %w", err)
}
if d <= 0 {
return 0, fmt.Errorf("start_timeout must be a positive duration, -1, or forever, got %q", s)
}
return d, nil
}
// effectiveStartTimeout maps a Go API zero value to DefaultStartTimeout.
func effectiveStartTimeout(d time.Duration) time.Duration {
if d == 0 {
return DefaultStartTimeout
}
return d
}
// effectivePythonRuntime returns the runtime string passed to the Python worker.
// When Runtime is empty: sync for WSGI, gevent for ESGI, uvloop for ASGI.
func effectivePythonRuntime(iface, runtime string) string {
if iface == "asgi" && runtime == "libuv" {
runtime = "uvloop" // legacy alias; configuration name is uvloop
}
if runtime != "" {
return runtime
}
if iface == "asgi" {
return "uvloop"
}
if iface == "esgi" {
return "gevent"
}
return "sync"
}
func validatePythonRuntime(iface, runtime string) error {
eff := effectivePythonRuntime(iface, runtime)
switch iface {
case "wsgi":
if eff != "sync" && eff != "gevent" {
return fmt.Errorf("wsgi runtime must be sync or gevent, got %q", eff)
}
case "esgi":
if eff != "gevent" {
return fmt.Errorf("esgi runtime must be gevent, got %q", eff)
}
case "asgi":
if eff != "native" && eff != "uvloop" {
return fmt.Errorf("asgi runtime must be native or uvloop, got %q", eff)
}
default:
return fmt.Errorf("unknown python interface %q", iface)
}
return nil
}
// UnmarshalCaddyfile implements caddyfile.Unmarshaler.
func (f *CaddySnake) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
for d.Next() {
args := d.RemainingArgs()
if len(args) == 1 {
f.ModuleWsgi = args[0]
} else if len(args) == 0 {
for nesting := d.Nesting(); d.NextBlock(nesting); {
switch d.Val() {
case "module_asgi":
if !d.Args(&f.ModuleAsgi) {
return d.Errf("expected exactly one argument for module_asgi")
}
case "module_esgi":
if !d.Args(&f.ModuleEsgi) {
return d.Errf("expected exactly one argument for module_esgi")
}
case "module_wsgi":
if !d.Args(&f.ModuleWsgi) {
return d.Errf("expected exactly one argument for module_wsgi")
}
case "runtime":
if !d.Args(&f.Runtime) {
return d.Errf("expected exactly one argument for runtime")
}
case "lifespan":
if !d.Args(&f.Lifespan) || (f.Lifespan != "on" && f.Lifespan != "off") {
return d.Errf("expected exactly one argument for lifespan: on|off")
}
case "working_dir":
if !d.Args(&f.WorkingDir) {
return d.Errf("expected exactly one argument for working_dir")
}
case "venv":
if !d.Args(&f.VenvPath) {
return d.Errf("expected exactly one argument for venv")
}
case "workers":
if !d.Args(&f.Workers) {
return d.Errf("expected exactly one argument for workers")
}
case "start_timeout":
if !d.Args(&f.StartTimeout) {
return d.Errf("expected exactly one argument for start_timeout")
}
if _, err := parseStartTimeout(f.StartTimeout); err != nil {
return d.Errf("%v", err)
}
case "autoreload":
f.Autoreload = "on"
case "python_path":
if !d.Args(&f.PythonPath) {
return d.Errf("expected exactly one argument for python_path")
}
case "env_file":
var path string
if !d.Args(&path) {
return d.Errf("expected exactly one argument for env_file")
}
f.EnvFiles = append(f.EnvFiles, path)
case "env_var":
var name, value string
if !d.Args(&name, &value) {
return d.Errf("expected exactly two arguments for env_var: VARNAME value")
}
if err := validateEnvVarName(name); err != nil {
return d.Errf("invalid env_var name %q: %v", name, err)
}
if f.EnvVars == nil {
f.EnvVars = make(map[string]string)
}
f.EnvVars[name] = value
default:
return d.Errf("unknown subdirective: %s", d.Val())
}
}
} else {
return d.ArgErr()
}
}
return nil
}
// CaddyModule returns the Caddy module information.
func (CaddySnake) CaddyModule() caddy.ModuleInfo {
return caddy.ModuleInfo{
ID: "http.handlers.python",
New: func() caddy.Module { return new(CaddySnake) },
}
}
// Provision sets up the module.
func (f *CaddySnake) Provision(ctx caddy.Context) error {
var err error
f.logger = ctx.Logger(f)
cs, err := startCacheServer()
if err != nil {
return fmt.Errorf("in-process cache: %w", err)
}
f.cacheSrv = cs
cacheAddr := cs.Addr()
success := false
defer func() {
if !success && f.cacheSrv != nil {
_ = f.cacheSrv.Close()
f.cacheSrv = nil
}
}()
workers, _ := strconv.Atoi(f.Workers)
if workers <= 0 {
workers = runtime.GOMAXPROCS(0)
}
startTimeout, err := parseStartTimeout(f.StartTimeout)
if err != nil {
return err
}
isDynamic := containsPlaceholder(f.ModuleWsgi) || containsPlaceholder(f.ModuleAsgi) ||
containsPlaceholder(f.ModuleEsgi) ||
containsPlaceholder(f.WorkingDir) || containsPlaceholder(f.VenvPath) ||
envFilesContainPlaceholder(f.EnvFiles) || envMapContainsPlaceholder(f.EnvVars)
if isDynamic {
if err = f.provisionDynamic(workers, cacheAddr, startTimeout); err != nil {
return err
}
success = true
return nil
}
pythonBin := resolvePythonInterpreter(f.PythonPath, f.VenvPath)
envFiles := cloneEnvFiles(f.EnvFiles)
envVars := cloneEnvVars(f.EnvVars)
if f.ModuleWsgi != "" {
rt := effectivePythonRuntime("wsgi", f.Runtime)
f.app, err = NewPythonWorkerGroup("wsgi", f.ModuleWsgi, f.WorkingDir, f.VenvPath, f.Lifespan, rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, f.logger)
if err != nil {
return err
}
if f.Lifespan != "" {
f.logger.Warn("lifespan attribute is ignored in WSGI mode", zap.String("lifespan", f.Lifespan))
}
f.logger.Info("serving wsgi app", zap.String("module_wsgi", f.ModuleWsgi), zap.String("working_dir", f.WorkingDir), zap.String("venv_path", f.VenvPath), zap.String("python", pythonBin), zap.String("runtime", rt))
} else if f.ModuleAsgi != "" {
rt := effectivePythonRuntime("asgi", f.Runtime)
f.app, err = NewPythonWorkerGroup("asgi", f.ModuleAsgi, f.WorkingDir, f.VenvPath, f.Lifespan, rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, f.logger)
if err != nil {
return err
}
f.logger.Info("serving asgi app", zap.String("module_asgi", f.ModuleAsgi), zap.String("working_dir", f.WorkingDir), zap.String("venv_path", f.VenvPath), zap.String("python", pythonBin), zap.String("runtime", rt))
} else if f.ModuleEsgi != "" {
rt := effectivePythonRuntime("esgi", f.Runtime)
f.app, err = NewPythonWorkerGroup("esgi", f.ModuleEsgi, f.WorkingDir, f.VenvPath, "", rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, f.logger)
if err != nil {
return err
}
if f.Lifespan != "" {
f.logger.Warn("lifespan is for ASGI only; ignored in ESGI mode", zap.String("lifespan", f.Lifespan))
}
f.logger.Info("serving esgi app", zap.String("module_esgi", f.ModuleEsgi), zap.String("working_dir", f.WorkingDir), zap.String("venv_path", f.VenvPath), zap.String("python", pythonBin), zap.String("runtime", rt))
} else {
return errors.New("a wsgi, asgi, or esgi app must be specified")
}
if f.Autoreload == "on" {
watchDir := f.WorkingDir
if watchDir == "" {
watchDir = "."
}
absDir, absErr := filepath.Abs(watchDir)
if absErr != nil {
return fmt.Errorf("autoreload: %w", absErr)
}
var factory func() (AppServer, error)
if f.ModuleWsgi != "" {
rt := effectivePythonRuntime("wsgi", f.Runtime)
factory = func() (AppServer, error) {
return NewPythonWorkerGroup("wsgi", f.ModuleWsgi, f.WorkingDir, f.VenvPath, f.Lifespan, rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, f.logger)
}
} else if f.ModuleAsgi != "" {
rt := effectivePythonRuntime("asgi", f.Runtime)
factory = func() (AppServer, error) {
return NewPythonWorkerGroup("asgi", f.ModuleAsgi, f.WorkingDir, f.VenvPath, f.Lifespan, rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, f.logger)
}
} else {
rt := effectivePythonRuntime("esgi", f.Runtime)
factory = func() (AppServer, error) {
return NewPythonWorkerGroup("esgi", f.ModuleEsgi, f.WorkingDir, f.VenvPath, "", rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, f.logger)
}
}
// Keep Caddy running on reload errors; failed app serves 503 until recovery.
f.app, err = NewAutoreloadableApp(f.app, absDir, factory, f.logger, nil)
if err != nil {
return fmt.Errorf("autoreload: %w", err)
}
}
success = true
return nil
}
// provisionDynamic sets up the module in dynamic mode where Caddy placeholders
// in module_wsgi/module_asgi/module_esgi, working_dir, or venv are resolved per-request.
func (f *CaddySnake) provisionDynamic(workers int, cacheAddr string, startTimeout time.Duration) error {
autoreload := f.Autoreload == "on"
pythonPath := f.PythonPath
envFilePatterns := cloneEnvFiles(f.EnvFiles)
envVarPatterns := cloneEnvVars(f.EnvVars)
logger := f.logger
if f.ModuleWsgi != "" {
lifespan := f.Lifespan
rt := effectivePythonRuntime("wsgi", f.Runtime)
factory := func(module, dir, venv string, envFiles []string, envVars map[string]string) (AppServer, error) {
pythonBin := resolvePythonInterpreter(pythonPath, venv)
return NewPythonWorkerGroup("wsgi", module, dir, venv, lifespan, rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, logger)
}
var err error
f.app, err = NewDynamicApp(f.ModuleWsgi, f.WorkingDir, f.VenvPath, envFilePatterns, envVarPatterns, factory, f.logger, autoreload, nil)
if err != nil {
return err
}
if f.Lifespan != "" {
f.logger.Warn("lifespan attribute is ignored in WSGI mode", zap.String("lifespan", f.Lifespan))
}
f.logger.Info("serving dynamic wsgi app",
zap.String("module_wsgi", f.ModuleWsgi),
zap.String("working_dir", f.WorkingDir),
zap.String("venv_path", f.VenvPath),
)
} else if f.ModuleAsgi != "" {
lifespan := f.Lifespan
rt := effectivePythonRuntime("asgi", f.Runtime)
factory := func(module, dir, venv string, envFiles []string, envVars map[string]string) (AppServer, error) {
pythonBin := resolvePythonInterpreter(pythonPath, venv)
return NewPythonWorkerGroup("asgi", module, dir, venv, lifespan, rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, logger)
}
var err error
f.app, err = NewDynamicApp(f.ModuleAsgi, f.WorkingDir, f.VenvPath, envFilePatterns, envVarPatterns, factory, f.logger, autoreload, nil)
if err != nil {
return err
}
f.logger.Info("serving dynamic asgi app",
zap.String("module_asgi", f.ModuleAsgi),
zap.String("working_dir", f.WorkingDir),
zap.String("venv_path", f.VenvPath),
)
} else if f.ModuleEsgi != "" {
rt := effectivePythonRuntime("esgi", f.Runtime)
factory := func(module, dir, venv string, envFiles []string, envVars map[string]string) (AppServer, error) {
pythonBin := resolvePythonInterpreter(pythonPath, venv)
return NewPythonWorkerGroup("esgi", module, dir, venv, "", rt, workers, pythonBin, cacheAddr, envFiles, envVars, startTimeout, logger)
}
var err error
f.app, err = NewDynamicApp(f.ModuleEsgi, f.WorkingDir, f.VenvPath, envFilePatterns, envVarPatterns, factory, f.logger, autoreload, nil)
if err != nil {
return err
}
if f.Lifespan != "" {
f.logger.Warn("lifespan is for ASGI only; ignored in dynamic ESGI mode", zap.String("lifespan", f.Lifespan))
}
f.logger.Info("serving dynamic esgi app",
zap.String("module_esgi", f.ModuleEsgi),
zap.String("working_dir", f.WorkingDir),
zap.String("venv_path", f.VenvPath),
)
} else {
return errors.New("a wsgi, asgi, or esgi app must be specified for dynamic mode")
}
return nil
}
// Validate implements caddy.Validator.
func (m *CaddySnake) Validate() error {
n := 0
if m.ModuleWsgi != "" {
n++
}
if m.ModuleAsgi != "" {
n++
}
if m.ModuleEsgi != "" {
n++
}
if n != 1 {
return errors.New("exactly one of module_wsgi, module_asgi, or module_esgi is required")
}
var iface string
switch {
case m.ModuleWsgi != "":
iface = "wsgi"
case m.ModuleAsgi != "":
iface = "asgi"
default:
iface = "esgi"
}
if err := validatePythonRuntime(iface, m.Runtime); err != nil {
return err
}
if m.Workers != "" {
w, err := strconv.Atoi(m.Workers)
if err != nil || w < 0 {
return fmt.Errorf("invalid workers value: %s", m.Workers)
}
}
if _, err := parseStartTimeout(m.StartTimeout); err != nil {
return err
}
if m.Lifespan != "" && m.Lifespan != "on" && m.Lifespan != "off" {
return fmt.Errorf("lifespan must be 'on' or 'off', got: %s", m.Lifespan)
}
if err := validateEnvVars(m.EnvVars); err != nil {
return err
}
return nil
}
// Cleanup frees resources uses by module
func (m *CaddySnake) Cleanup() error {
var err error
if m != nil && m.app != nil {
m.logger.Info("cleaning up module")
err = m.app.Cleanup()
}
if m != nil && m.cacheSrv != nil {
if cerr := m.cacheSrv.Close(); cerr != nil && err == nil {
err = cerr
}
m.cacheSrv = nil
}
return err
}
// ServeHTTP implements caddyhttp.MiddlewareHandler.
func (f CaddySnake) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyhttp.Handler) error {
if err := f.app.HandleRequest(w, r); err != nil {
return err
}
return nil
}
// Interface guards
var (
_ caddy.Provisioner = (*CaddySnake)(nil)
_ caddy.Validator = (*CaddySnake)(nil)
_ caddy.CleanerUpper = (*CaddySnake)(nil)
_ caddyhttp.MiddlewareHandler = (*CaddySnake)(nil)
_ caddyfile.Unmarshaler = (*CaddySnake)(nil)
)
func parsePythonDirective(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error) {
var app CaddySnake
if err := app.UnmarshalCaddyfile(h.Dispenser); err != nil {
return nil, err
}
return app, nil
}
// resolvePythonInterpreter determines the Python interpreter to use.
// Priority: explicit python_path > venv/bin/python > system python3.
func resolvePythonInterpreter(pythonPath, venvPath string) string {
if pythonPath != "" {
return pythonPath
}
if venvPath != "" {
var binDir string
if runtime.GOOS == "windows" {
binDir = "Scripts"
} else {
binDir = "bin"
}
venvPython := filepath.Join(venvPath, binDir, "python3")
if runtime.GOOS == "windows" {
venvPython = filepath.Join(venvPath, binDir, "python.exe")
}
if _, err := os.Stat(venvPython); err == nil {
return venvPython
}
venvPython2 := filepath.Join(venvPath, binDir, "python")
if _, err := os.Stat(venvPython2); err == nil {
return venvPython2
}
}
return "python3"
}
// writeCaddysnakePyBundle writes the embedded worker implementation as worker_main.py
// inside a private temp directory so user apps can import the PyPI `caddysnake` package.
func writeCaddysnakePyBundle() (scriptPath, bundleDir string, err error) {
bundleDir, err = os.MkdirTemp("", "caddysnake-worker-*")
if err != nil {
return "", "", err
}
if runtime.GOOS != "windows" {
if chErr := os.Chmod(bundleDir, 0700); chErr != nil {
os.RemoveAll(bundleDir)
return "", "", chErr
}
}
scriptPath = filepath.Join(bundleDir, "worker_main.py")
if wrErr := os.WriteFile(scriptPath, []byte(caddysnake_py), 0600); wrErr != nil {
os.RemoveAll(bundleDir)
return "", "", wrErr
}
return scriptPath, bundleDir, nil
}
// proxyBufferPool implements httputil.BufferPool using sync.Pool to reduce GC pressure.
type proxyBufferPool struct {
pool sync.Pool
}
func (p *proxyBufferPool) Get() []byte {
b := p.pool.Get()
if b == nil {
return make([]byte, 32*1024)
}
return *b.(*[]byte)
}
func (p *proxyBufferPool) Put(b []byte) {
p.pool.Put(&b)
}
var sharedProxyBufferPool = &proxyBufferPool{}
type PythonWorker struct {
Interface string
App string
WorkingDir string
Venv string
Lifespan string
Runtime string
PythonBin string
Socket *os.File
SockDir string // private directory containing the socket (Unix only)
ScriptPath string
DialNet string // "unix" or "tcp"
DialAddr string // socket path or host:port
CacheAddr string // CADDYSNAKE_CACHE_ADDR: unix://path (Unix) or 127.0.0.1:port (Windows); empty = omit env
WorkerID string // CADDYSNAKE_WORKER_ID: stable index 0..N-1 within the worker group
EnvFiles []string
EnvVars map[string]string
StartTimeout time.Duration // 0 = DefaultStartTimeout; <0 = indefinite
logger *zap.Logger
Cmd *exec.Cmd
cmdWaitCh chan error // receives Cmd.Wait result; only Wait once
cmdReaped bool
Transport *http.Transport
Proxy *httputil.ReverseProxy
}
func NewPythonWorker(iface, app, workingDir, venv, lifespan, pyRuntime, pythonBin, scriptPath, cacheAddr, workerID string, envFiles []string, envVars map[string]string, startTimeout time.Duration, logger *zap.Logger) (*PythonWorker, error) {
var socket *os.File
var sockDir string
var err error
if runtime.GOOS == "windows" {
socket, err = os.CreateTemp("", "caddysnake-worker.port*")
} else {
sockDir, err = os.MkdirTemp("", "caddysnake-*")
if err != nil {
return nil, err
}
if chErr := os.Chmod(sockDir, 0700); chErr != nil {
os.RemoveAll(sockDir)
return nil, chErr
}
socket, err = os.Create(filepath.Join(sockDir, "worker.sock"))
}
if err != nil {
if sockDir != "" {
os.RemoveAll(sockDir)
}
return nil, err
}
path := socket.Name()
socket.Close()
dialNet := "unix"
dialAddr := path
if runtime.GOOS == "windows" {
dialNet = "tcp"
dialAddr = "" // set after Python writes port to path
}
w := &PythonWorker{
Interface: iface,
App: app,
WorkingDir: workingDir,
Venv: venv,
Lifespan: lifespan,
Runtime: pyRuntime,
PythonBin: pythonBin,
Socket: socket,
SockDir: sockDir,
ScriptPath: scriptPath,
DialNet: dialNet,
DialAddr: dialAddr,
CacheAddr: cacheAddr,
WorkerID: workerID,
EnvFiles: cloneEnvFiles(envFiles),
EnvVars: cloneEnvVars(envVars),
StartTimeout: startTimeout,
logger: logger,
}
err = w.Start()
return w, err
}
func (w *PythonWorker) Start() error {
w.Transport = &http.Transport{
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
return w.dialWithRetry(ctx)
},
MaxIdleConns: 1024,
MaxIdleConnsPerHost: 256,
IdleConnTimeout: 90 * time.Second,
DisableCompression: true,
}
w.Proxy = &httputil.ReverseProxy{
Rewrite: func(req *httputil.ProxyRequest) {
setPythonWorkerOutboundHeaders(req, w.DialAddr)
},
Transport: w.Transport,
BufferPool: sharedProxyBufferPool,
}
workingDir := w.WorkingDir
if workingDir == "" {
workingDir, _ = os.Getwd()
}
args := []string{
w.ScriptPath,
"--interface", w.Interface,
"--app", w.App,
"--socket", w.Socket.Name(),
}
if workingDir != "" {
args = append(args, "--working-dir", workingDir)
}
if w.Venv != "" {
args = append(args, "--venv", w.Venv)
}
if w.Lifespan != "" {
args = append(args, "--lifespan", w.Lifespan)
}
if w.Runtime != "" {
args = append(args, "--runtime", w.Runtime)
}
w.Cmd = exec.Command(w.PythonBin, args...)
w.Cmd.Stdout = os.Stdout
w.Cmd.Stderr = os.Stderr
fileVars, err := loadEnvFiles(w.WorkingDir, w.EnvFiles)
if err != nil {
return err
}
w.Cmd.Env = buildWorkerEnv(os.Environ(), fileVars, w.EnvVars, workerInternalEnv(w.Interface, w.CacheAddr, w.WorkerID)...)
setSysProcAttr(w.Cmd)
if err := w.Cmd.Start(); err != nil {
return err
}
w.cmdWaitCh = make(chan error, 1)
go func() {
w.cmdWaitCh <- w.Cmd.Wait()
}()
timeout := effectiveStartTimeout(w.StartTimeout)
if runtime.GOOS == "windows" {
port, err := waitForPortFile(w.Socket.Name(), timeout, w.cmdWaitCh, w.logger)
if err != nil {
w.reapWorkerAfterStartFailure(err)
return fmt.Errorf("waiting for Python worker port file: %w", err)
}
w.DialAddr = "127.0.0.1:" + strconv.Itoa(port)
} else if err := waitForUnixSocket(w.Socket.Name(), timeout, w.cmdWaitCh, w.logger); err != nil {
w.reapWorkerAfterStartFailure(err)
return fmt.Errorf("waiting for Python worker socket: %w", err)
}
return nil
}
// reapWorkerAfterStartFailure ensures the worker process is reaped exactly once
// after a failed readiness wait.
func (w *PythonWorker) reapWorkerAfterStartFailure(err error) {
if w.cmdReaped {
return
}
if errors.Is(err, errWorkerExited) {
// cmdWaitCh was already consumed by the readiness wait.
w.cmdReaped = true
return
}
if w.Cmd != nil && w.Cmd.Process != nil {
_ = w.Cmd.Process.Kill()
}
if w.cmdWaitCh != nil {
<-w.cmdWaitCh
}
w.cmdReaped = true
}
// maybeWarnSlowStart logs once when the app is still loading past startTimeoutWarnAfter
// and the configured timeout allows waiting longer than that.
func maybeWarnSlowStart(logger *zap.Logger, warned *bool, start time.Time, timeout time.Duration, path, kind string) {
if *warned || logger == nil {
return
}
hasDeadline := timeout >= 0
if hasDeadline && timeout <= startTimeoutWarnAfter {
return
}
if time.Since(start) < startTimeoutWarnAfter {
return
}
logger.Warn("Python app is taking a long time to load; still waiting for worker to become ready",
zap.String(kind, path),
zap.Duration("waited", time.Since(start)),
)
*warned = true
}
func checkWorkerExited(exited <-chan error, path, kind string) error {
if exited == nil {
return nil
}
select {
case err := <-exited:
if err != nil {
return fmt.Errorf("%w (%s %s): %w", errWorkerExited, kind, path, err)
}
return fmt.Errorf("%w (%s %s)", errWorkerExited, kind, path)
default:
return nil
}
}
// waitForPortFile polls the given file path until it contains a valid port number.
// timeout < 0 means wait indefinitely. exited, if non-nil, fails fast when the
// worker process exits. logger may be nil.
func waitForPortFile(path string, timeout time.Duration, exited <-chan error, logger *zap.Logger) (int, error) {
start := time.Now()
var deadline time.Time
hasDeadline := timeout >= 0
if hasDeadline {
deadline = start.Add(timeout)
}
warned := false
for {
if err := checkWorkerExited(exited, path, "port file"); err != nil {
return 0, err
}
data, err := os.ReadFile(path)
if err == nil && len(data) > 0 {
port, err := strconv.Atoi(strings.TrimSpace(string(data)))
if err == nil && port > 0 && port < 65536 {
return port, nil
}
}
if hasDeadline && !time.Now().Before(deadline) {
return 0, fmt.Errorf("port file %s not ready within %v", path, timeout)
}
maybeWarnSlowStart(logger, &warned, start, timeout, path, "port_file")
time.Sleep(50 * time.Millisecond)
}
}
// waitForUnixSocket polls until a unix socket at path accepts connections.
// timeout < 0 means wait indefinitely. exited, if non-nil, fails fast when the
// worker process exits. logger may be nil.
func waitForUnixSocket(path string, timeout time.Duration, exited <-chan error, logger *zap.Logger) error {
start := time.Now()
var deadline time.Time
hasDeadline := timeout >= 0
if hasDeadline {
deadline = start.Add(timeout)
}
warned := false
var dialer net.Dialer
for {
if err := checkWorkerExited(exited, path, "unix socket"); err != nil {
return err
}
conn, err := dialer.DialContext(context.Background(), "unix", path)
if err == nil {
conn.Close()
return nil
}
if hasDeadline && !time.Now().Before(deadline) {
return fmt.Errorf("unix socket %s not ready within %v", path, timeout)
}
maybeWarnSlowStart(logger, &warned, start, timeout, path, "socket")
time.Sleep(50 * time.Millisecond)
}
}
// dialWithRetry attempts to establish a connection with retry logic
func (w *PythonWorker) dialWithRetry(ctx context.Context) (net.Conn, error) {
const maxRetries = 5
const baseDelay = 100 * time.Millisecond
var dialer net.Dialer
for attempt := 0; attempt < maxRetries; attempt++ {
conn, err := dialer.DialContext(ctx, w.DialNet, w.DialAddr)
if err == nil {
return conn, nil
}
if attempt == maxRetries-1 {
return nil, fmt.Errorf("failed to connect after %d attempts: %w", maxRetries, err)
}
delay := baseDelay * time.Duration(1<<attempt)
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(delay):
}
}
return nil, fmt.Errorf("unexpected error in dialWithRetry")
}
func (w *PythonWorker) Cleanup() error {
if w.Transport != nil {
w.Transport.CloseIdleConnections()
}
if w.Cmd != nil && w.Cmd.Process != nil && !w.cmdReaped {
// On Windows, Signal(SIGTERM) is not supported; only Kill works.
// Send SIGTERM on Unix for graceful shutdown (ASGI lifespan), Kill on Windows.
if runtime.GOOS == "windows" {
_ = w.Cmd.Process.Kill()
} else {
_ = w.Cmd.Process.Signal(syscall.SIGTERM)
}
if w.cmdWaitCh != nil {
select {
case <-w.cmdWaitCh:
w.cmdReaped = true
case <-time.After(5 * time.Second):
_ = w.Cmd.Process.Kill()
<-w.cmdWaitCh
w.cmdReaped = true
}
} else {
done := make(chan error, 1)
go func() {
_, err := w.Cmd.Process.Wait()
done <- err
}()
select {
case <-done:
w.cmdReaped = true
case <-time.After(5 * time.Second):
_ = w.Cmd.Process.Kill()
<-done
w.cmdReaped = true
}
}
}
if w.Socket != nil {
w.Socket.Close()
os.Remove(w.Socket.Name())
if w.SockDir != "" {
os.RemoveAll(w.SockDir)
}
}
return nil
}
func (w *PythonWorker) HandleRequest(rw http.ResponseWriter, req *http.Request) error {
w.Proxy.ServeHTTP(rw, req)
return nil
}
type PythonWorkerGroup struct {
Workers []*PythonWorker
roundRobin atomic.Uint64
BundleDir string
ScriptPath string // path to worker_main.py inside BundleDir
}
func NewPythonWorkerGroup(iface, app, workingDir, venv, lifespan, runtime string, count int, pythonBin, cacheAddr string, envFiles []string, envVars map[string]string, startTimeout time.Duration, logger *zap.Logger) (*PythonWorkerGroup, error) {
scriptPath, bundleDir, err := writeCaddysnakePyBundle()
if err != nil {
return nil, fmt.Errorf("failed to write worker bundle: %w", err)
}
errs := make([]error, count)
workers := make([]*PythonWorker, count)
for i := 0; i < count; i++ {
workers[i], errs[i] = NewPythonWorker(iface, app, workingDir, venv, lifespan, runtime, pythonBin, scriptPath, cacheAddr, strconv.Itoa(i), envFiles, envVars, startTimeout, logger)
}
wg := &PythonWorkerGroup{
Workers: workers,
BundleDir: bundleDir,
ScriptPath: scriptPath,
}
if err := errors.Join(errs...); err != nil {
// Prefer the start/readiness error; discard cleanup noise (e.g. ECHILD).
_ = wg.Cleanup()
return nil, err
}
return wg, nil
}
func (wg *PythonWorkerGroup) Cleanup() error {
if wg == nil {
return nil
}
errs := make([]error, len(wg.Workers))
for i, worker := range wg.Workers {
if worker != nil {
errs[i] = worker.Cleanup()
}
}
if wg.BundleDir != "" {
_ = os.RemoveAll(wg.BundleDir)
}
return errors.Join(errs...)
}
func (wg *PythonWorkerGroup) HandleRequest(rw http.ResponseWriter, req *http.Request) error {
n := wg.roundRobin.Add(1)
idx := int(n % uint64(len(wg.Workers)))
return wg.Workers[idx].HandleRequest(rw, req)
}
func init() {
applyStartTimeoutWarnAfterEnv()
caddy.RegisterModule(CaddySnake{})
httpcaddyfile.RegisterHandlerDirective("python", parsePythonDirective)
httpcaddyfile.RegisterDirectiveOrder("python", httpcaddyfile.Before, "route")
caddycmd.RegisterCommand(caddycmd.Command{
Name: "python-server",
Usage: "--server-type wsgi|asgi|esgi --app <module> " +
"[--domain <example.com>] [--listen <addr>] [--workers <count>] " +
"[--python-path <path>] [--working-dir <path>] [--venv <path>] " +
"[--env-file <path>] [--env-var NAME=VALUE] " +
"[--start-timeout=<duration|-1|forever>] " +
"[--static-path <path>] [--static-route <route>] " +
"[--runtime <name>] [--lifespan on|off] " +
"[--debug] [--access-logs] [--autoreload]",
Short: "Spins up a Python server",
Long: `
A Python WSGI, ASGI, or ESGI server designed for apps and frameworks.
Python-block options mirror the Caddyfile python directive (workers, venv,
working_dir, env_file, env_var, start_timeout, runtime, lifespan, autoreload,
python_path). CLI-only flags cover listen address, HTTPS domain, and static files.
For an indefinite readiness wait use --start-timeout=-1 (equals form) or
--start-timeout forever. Do not pass a bare "-1" as a separate argv token;
Cobra/pflag treats it as a flag name.
You can specify a custom socket address using the '--listen' option. You can also specify the number of workers to spawn.
Providing a domain name with the '--domain' flag enables HTTPS and sets the listener to the appropriate secure port.
Ensure DNS A/AAAA records are correctly set up if using a public domain for secure connections.
`,
CobraFunc: func(cmd *cobra.Command) {
cmd.Flags().StringP("server-type", "t", "", "Required. The type of server to use: wsgi|asgi|esgi")
cmd.Flags().StringP("app", "a", "", "Required. App module to be imported")
cmd.Flags().StringP("domain", "d", "", "Domain name at which to serve the files")
cmd.Flags().StringP("listen", "l", "", "The address to which to bind the listener")
cmd.Flags().StringP("workers", "w", "0", "The number of workers to spawn")
cmd.Flags().String("python-path", "", "Path to the Python interpreter")
cmd.Flags().String("working-dir", "", "Working directory for the Python app")
cmd.Flags().String("venv", "", "Path to a Python virtual environment to use")
cmd.Flags().StringSlice("env-file", nil, "Dotenv file loaded into worker env (repeatable; later files override)")
cmd.Flags().StringSlice("env-var", nil, "Inline worker env var as NAME=VALUE (repeatable; overrides env-file)")
cmd.Flags().String("start-timeout", "", "Wait for worker readiness (default: 120s; use =-1 or forever for indefinite)")
cmd.Flags().String("static-path", "", "Path to a static directory to serve: path/to/static")
cmd.Flags().String("static-route", "/static", "Route to serve the static directory: /static")
cmd.Flags().Bool("debug", false, "Enable debug logs")
cmd.Flags().Bool("access-logs", false, "Enable access logs")
cmd.Flags().Bool("autoreload", false, "Watch .py files and reload on changes")
cmd.Flags().String("lifespan", "off", "Enable ASGI lifespan support (ignored in WSGI mode)")
cmd.Flags().String("runtime", "", "Worker runtime (wsgi: sync|gevent; esgi: gevent only; asgi: native|uvloop); defaults: sync for WSGI, gevent for ESGI, uvloop for ASGI")
cmd.RunE = caddycmd.WrapCommandFuncForCobra(pythonServer)
},
})
}
// pythonServer is inspired on the php-server command of the Frankenphp project (MIT License)
func pythonServer(fs caddycmd.Flags) (int, error) {
caddy.TrapSignals()
domain := fs.String("domain")
app := fs.String("app")
listen := fs.String("listen")
workers := fs.String("workers")
debug := fs.Bool("debug")
accessLogs := fs.Bool("access-logs")
autoreload := fs.Bool("autoreload")
staticPath := fs.String("static-path")
staticRoute := fs.String("static-route")
serverType := fs.String("server-type")
pythonPath := fs.String("python-path")
workingDir := fs.String("working-dir")
venv := fs.String("venv")
lifespan := fs.String("lifespan")
runtimeFlag := fs.String("runtime")
startTimeout := fs.String("start-timeout")
envFiles, err := fs.GetStringSlice("env-file")
if err != nil {
return caddy.ExitCodeFailedStartup, err
}
envVarFlags, err := fs.GetStringSlice("env-var")
if err != nil {
return caddy.ExitCodeFailedStartup, err
}
envVars, err := parseCLIEnvVars(envVarFlags)
if err != nil {
return caddy.ExitCodeFailedStartup, err
}
if serverType == "" {
return caddy.ExitCodeFailedStartup, errors.New("--server-type is required")
}
if serverType != "wsgi" && serverType != "asgi" && serverType != "esgi" {
return caddy.ExitCodeFailedStartup, fmt.Errorf("invalid --server-type %q (want wsgi, asgi, or esgi)", serverType)
}
if app == "" {
return caddy.ExitCodeFailedStartup, errors.New("--app is required")
}
gzip, err := caddy.GetModule("http.encoders.gzip")
if err != nil {
return caddy.ExitCodeFailedStartup, err
}
zstd, err := caddy.GetModule("http.encoders.zstd")
if err != nil {
return caddy.ExitCodeFailedStartup, err
}
encodings := caddy.ModuleMap{
"zstd": caddyconfig.JSON(zstd.New(), nil),
"gzip": caddyconfig.JSON(gzip.New(), nil),
}
prefer := []string{"zstd", "gzip"}
pythonHandler := CaddySnake{}
if serverType == "wsgi" {
pythonHandler.ModuleWsgi = app
} else if serverType == "asgi" {
pythonHandler.ModuleAsgi = app
} else {
pythonHandler.ModuleEsgi = app
}
if venv != "" {
pythonHandler.VenvPath = venv
} else if venv := os.Getenv("VIRTUAL_ENV"); venv != "" {
pythonHandler.VenvPath = venv
}
pythonHandler.Workers = workers
pythonHandler.PythonPath = pythonPath
if autoreload {
pythonHandler.Autoreload = "on"
}
pythonHandler.WorkingDir = workingDir
pythonHandler.Lifespan = lifespan
pythonHandler.Runtime = runtimeFlag
pythonHandler.StartTimeout = startTimeout
pythonHandler.EnvFiles = cloneEnvFiles(envFiles)
pythonHandler.EnvVars = envVars
if err := pythonHandler.Validate(); err != nil {
return caddy.ExitCodeFailedStartup, err
}
routes := caddyhttp.RouteList{}
if staticPath != "" {
if strings.HasSuffix(staticRoute, "/") {
staticRoute = staticRoute + "*"
} else if !strings.HasSuffix(staticRoute, "/*") {
staticRoute = staticRoute + "/*"
}
staticRoute := caddyhttp.Route{
MatcherSetsRaw: []caddy.ModuleMap{
{
"path": caddyconfig.JSON(caddyhttp.MatchPath{staticRoute}, nil),
},
},
HandlersRaw: []json.RawMessage{
caddyconfig.JSONModuleObject(encode.Encode{
EncodingsRaw: encodings,
Prefer: prefer,
}, "handler", "encode", nil),
caddyconfig.JSON(map[string]interface{}{
"handler": "file_server",
"root": staticPath,
}, nil),
},
}
routes = append(routes, staticRoute)
}
mainRoute := caddyhttp.Route{
MatcherSetsRaw: []caddy.ModuleMap{
{
"path": caddyconfig.JSON(caddyhttp.MatchPath{"/*"}, nil),
},
},
HandlersRaw: []json.RawMessage{
caddyconfig.JSONModuleObject(encode.Encode{
EncodingsRaw: encodings,
Prefer: prefer,
}, "handler", "encode", nil),
caddyconfig.JSONModuleObject(pythonHandler, "handler", "python", nil),
},
}
routes = append(routes, mainRoute)
subroute := caddyhttp.Subroute{
Routes: routes,
}
route := caddyhttp.Route{
HandlersRaw: []json.RawMessage{caddyconfig.JSONModuleObject(subroute, "handler", "subroute", nil)},
}
if domain != "" {
route.MatcherSetsRaw = []caddy.ModuleMap{
{
"host": caddyconfig.JSON(caddyhttp.MatchHost{domain}, nil),
},
}
}
server := &caddyhttp.Server{
ReadHeaderTimeout: caddy.Duration(10 * time.Second),
IdleTimeout: caddy.Duration(30 * time.Second),
MaxHeaderBytes: 1024 * 10,
Routes: caddyhttp.RouteList{route},
}
if listen == "" {
if domain == "" {
listen = ":9080"
} else {
listen = ":" + strconv.Itoa(certmagic.HTTPSPort)
}
}
server.Listen = []string{listen}
if accessLogs {
server.Logs = &caddyhttp.ServerLogConfig{}
}
httpApp := caddyhttp.App{
Servers: map[string]*caddyhttp.Server{"srv0": server},
}
var f bool
cfg := &caddy.Config{
Admin: &caddy.AdminConfig{
Disabled: false,
Config: &caddy.ConfigSettings{
Persist: &f,
},
},
AppsRaw: caddy.ModuleMap{
"http": caddyconfig.JSON(httpApp, nil),
},
}
if debug {
cfg.Logging = &caddy.Logging{
Logs: map[string]*caddy.CustomLog{
"default": {
BaseLog: caddy.BaseLog{Level: zapcore.DebugLevel.CapitalString()},
},
},
}
}
if err := caddy.Run(cfg); err != nil {
return caddy.ExitCodeFailedStartup, err
}
log.Printf("Serving Python app on %s", listen)
select {}
}
// findSitePackagesInVenv searches for the site-packages directory in a given venv.
func findSitePackagesInVenv(venvPath string) (string, error) {
var sitePackagesPath string
if runtime.GOOS == "windows" {
sitePackagesPath = filepath.Join(venvPath, "Lib\\site-packages")
} else {
libPath := filepath.Join(venvPath, "lib")
pythonDir, err := findPythonDirectory(libPath)
if err != nil {
return "", err
}
sitePackagesPath = filepath.Join(libPath, pythonDir, "site-packages")
}
fileInfo, err := os.Stat(sitePackagesPath)
if err != nil {
if os.IsNotExist(err) {
return "", fmt.Errorf("site-packages directory does not exist in: %s", sitePackagesPath)
}
return "", err
}
if !fileInfo.IsDir() {
return "", fmt.Errorf("found site-packages is not a directory: %s", sitePackagesPath)
}
return sitePackagesPath, nil
}
// findWorkingDirectory checks if the directory exists and returns the absolute path
func findWorkingDirectory(workingDir string) (string, error) {
workingDirAbs, err := filepath.Abs(workingDir)
if err != nil {
return "", err
}
fileInfo, err := os.Stat(workingDirAbs)
if err != nil {
if os.IsNotExist(err) {
return "", fmt.Errorf("working_dir directory does not exist in: %s", workingDirAbs)
}
return "", err
}
if !fileInfo.IsDir() {
return "", fmt.Errorf("working_dir is not a directory: %s", workingDirAbs)
}
return workingDirAbs, nil
}
// findPythonDirectory searches for a directory that matches "python3.*" inside the given libPath.
func findPythonDirectory(libPath string) (string, error) {
entries, err := os.ReadDir(libPath)
if err != nil {
return "", fmt.Errorf("unable to read venv lib directory: %w", err)
}
for _, e := range entries {
if e.IsDir() && strings.HasPrefix(e.Name(), "python3") {
return e.Name(), nil
}
}
return "", errors.New("unable to find a python3.* directory in the venv")
}
package caddysnake
import (
"errors"
"fmt"
"net/http"
"os"
"path/filepath"
"regexp"
"sort"
"strings"
"sync"
"time"
"github.com/caddyserver/caddy/v2"
"github.com/fsnotify/fsnotify"
"go.uber.org/zap"
)
var validModulePattern = regexp.MustCompile(`^[a-zA-Z_][\w.]*:[a-zA-Z_]\w*$`)
// hasDotDotSegment reports whether the raw (pre-normalization) path contains
// a ".." segment. Checking before filepath.Abs/Clean is important: those
// normalize traversal sequences away (e.g. "/srv/apps/../../etc" becomes
// "/etc"), which would let placeholder-injected values escape the intended
// directory while passing a check on the normalized result.
func hasDotDotSegment(path string) bool {
for _, seg := range strings.FieldsFunc(path, func(r rune) bool {
return r == '/' || r == '\\'
}) {
if seg == ".." {
return true
}
}
return false
}
func validateResolvedValues(module, dir, venv string) error {
if !validModulePattern.MatchString(module) {
return fmt.Errorf("invalid module name: %q", module)
}
if dir != "" {
if hasDotDotSegment(dir) {
return fmt.Errorf("working directory contains path traversal: %q", dir)
}
if _, err := filepath.Abs(dir); err != nil {
return fmt.Errorf("invalid working directory: %w", err)
}
}
if venv != "" {
if hasDotDotSegment(venv) {
return fmt.Errorf("venv path contains path traversal: %q", venv)
}
if _, err := filepath.Abs(venv); err != nil {
return fmt.Errorf("invalid venv path: %w", err)
}
}
return nil
}
// containsPlaceholder checks if a string contains Caddy placeholders (e.g. {host.labels.0}).
func containsPlaceholder(s string) bool {
return strings.Contains(s, "{") && strings.Contains(s, "}")
}
func validateResolvedEnvConfig(dir string, envFiles []string) error {
for _, p := range envFiles {
if p == "" {
continue
}
if containsPlaceholder(p) {
return fmt.Errorf("env_file path contains unresolved placeholder: %q", p)
}
if hasDotDotSegment(p) {
return fmt.Errorf("env_file path contains path traversal: %q", p)
}
abs, err := resolveEnvFilePath(dir, p)
if err != nil {
return err
}
info, err := os.Stat(abs)
if err != nil {
return fmt.Errorf("env_file %q: %w", abs, err)
}
if info.IsDir() {
return fmt.Errorf("env_file %q is a directory", abs)
}
}
return nil
}
func envVarsCacheKey(m map[string]string) string {
if len(m) == 0 {
return ""
}
keys := make([]string, 0, len(m))
for k := range m {
keys = append(keys, k)
}
sort.Strings(keys)
var b strings.Builder
for _, k := range keys {
b.WriteString(k)
b.WriteByte('=')
b.WriteString(m[k])
b.WriteByte(';')
}
return b.String()
}
// appFactory is a function that creates a new AppServer for a resolved
// module, working directory, venv path, and env configuration.
type appFactory func(resolvedModule, resolvedDir, resolvedVenv string, envFiles []string, envVars map[string]string) (AppServer, error)
// appCreate tracks an in-flight factory call so concurrent requests for the
// same key wait for one create, while other keys are not blocked.
type appCreate struct {
done chan struct{}
app AppServer
err error
}
// DynamicApp implements AppServer by lazily importing Python apps based on
// Caddy placeholders resolved at request time. For example, when working_dir
// contains {host.labels.2}, each subdomain gets its own Python app instance
// imported from the corresponding directory.
type DynamicApp struct {
mu sync.RWMutex
apps map[string]AppServer
inflight map[string]*appCreate
closed bool
modulePattern string
workingDir string
venvPath string
envFilePatterns []string
envVarPatterns map[string]string
factory appFactory
logger *zap.Logger
// Autoreload fields
autoreload bool
watcher *fsnotify.Watcher
dirToKeys map[string][]string // abs working dir -> cache keys that use it
stopCh chan struct{}
exitOnReloadFailure func(code int) // if set and autoreload, process exits when app creation fails
}
// NewDynamicApp creates a DynamicApp that resolves placeholders from
// modulePattern, workingDir, venvPath, env_file paths, and env_var values at
// request time and lazily creates Python app instances via the supplied factory.
// When autoreload is true, if exitOnReloadFailure is non-nil it is called with
// code 1 when app creation fails (e.g. app deleted), so the process can terminate.
func NewDynamicApp(modulePattern, workingDir, venvPath string, envFilePatterns []string, envVarPatterns map[string]string, factory appFactory, logger *zap.Logger, autoreload bool, exitOnReloadFailure func(code int)) (*DynamicApp, error) {
d := &DynamicApp{
apps: make(map[string]AppServer),
inflight: make(map[string]*appCreate),
modulePattern: modulePattern,
workingDir: workingDir,
venvPath: venvPath,
envFilePatterns: cloneEnvFiles(envFilePatterns),
envVarPatterns: cloneEnvVars(envVarPatterns),
factory: factory,
logger: logger,
autoreload: autoreload,
exitOnReloadFailure: exitOnReloadFailure,
}
if autoreload {
watcher, err := fsnotify.NewWatcher()
if err != nil {
return nil, err
}
d.watcher = watcher
d.dirToKeys = make(map[string][]string)
d.stopCh = make(chan struct{})
go d.watchForChanges()
logger.Info("autoreload enabled for dynamic app")
}
return d, nil
}
// resolve uses the Caddy replacer from the request context to substitute
// placeholders in the module pattern, working directory, venv path, env files,
// and env_var values.
func (d *DynamicApp) resolve(r *http.Request) (key, module, dir, venv string, envFiles []string, envVars map[string]string) {
module = d.modulePattern
dir = d.workingDir
venv = d.venvPath
envFiles = cloneEnvFiles(d.envFilePatterns)
envVars = cloneEnvVars(d.envVarPatterns)
if repl, ok := r.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer); ok && repl != nil {
module = repl.ReplaceAll(module, "")
dir = repl.ReplaceAll(dir, "")
venv = repl.ReplaceAll(venv, "")
for i, p := range envFiles {
envFiles[i] = repl.ReplaceAll(p, "")
}
for name, value := range envVars {
envVars[name] = repl.ReplaceAll(value, "")
}
}
key = module + "|" + dir + "|" + venv + "|" + strings.Join(envFiles, ",") + "|" + envVarsCacheKey(envVars)
return
}
// getOrCreateApp returns an existing app for the given key, or creates one
// using the factory if it doesn't exist yet. The factory runs outside the
// exclusive lock so a slow or indefinite start_timeout for one tenant cannot
// stall other dynamic keys (same pattern as AutoreloadableApp.reload).
func (d *DynamicApp) getOrCreateApp(key, module, dir, venv string, envFiles []string, envVars map[string]string) (AppServer, error) {
if err := validateResolvedValues(module, dir, venv); err != nil {
return nil, err
}
if err := validateResolvedEnvConfig(dir, envFiles); err != nil {
return nil, err
}
for name, value := range envVars {
if containsPlaceholder(value) {
return nil, fmt.Errorf("env_var %q contains unresolved placeholder", name)
}
}
d.mu.RLock()
if d.closed {
d.mu.RUnlock()
return nil, errors.New("dynamic app shutting down")
}
app, ok := d.apps[key]
d.mu.RUnlock()
if ok {
return app, nil
}
d.mu.Lock()
if d.closed {
d.mu.Unlock()
return nil, errors.New("dynamic app shutting down")
}
app, ok = d.apps[key]
if ok {
d.mu.Unlock()
return app, nil
}
if c, creating := d.inflight[key]; creating {
d.mu.Unlock()
<-c.done
return c.app, c.err
}
c := &appCreate{done: make(chan struct{})}
d.inflight[key] = c
d.mu.Unlock()
d.logger.Info("dynamically importing python app",
zap.String("module", module),
zap.String("working_dir", dir),
zap.String("venv", venv),
)
// Factory runs outside the lock; if it panics, still remove the inflight
// entry and close done so waiters and Cleanup cannot hang forever.
var err error
func() {
defer func() {
if r := recover(); r != nil {
d.mu.Lock()
delete(d.inflight, key)
c.app = nil
c.err = fmt.Errorf("panic creating dynamic app: %v", r)
close(c.done)
d.mu.Unlock()
panic(r)
}
}()
app, err = d.factory(module, dir, venv, cloneEnvFiles(envFiles), cloneEnvVars(envVars))
}()
d.mu.Lock()
delete(d.inflight, key)
if d.closed {
d.mu.Unlock()
if app != nil {
_ = app.Cleanup()
}
if err == nil {
err = errors.New("dynamic app shutting down")
}
c.app = nil
c.err = err
close(c.done)
return nil, err
}
if err == nil {
d.apps[key] = app
if d.autoreload && dir != "" {
d.startWatchingDir(dir, key)
}
}
c.app = app
c.err = err
close(c.done)
d.mu.Unlock()
return app, err
}
func (d *DynamicApp) startWatchingDir(dir, key string) {
absDir, err := filepath.Abs(dir)
if err != nil {
d.logger.Warn("autoreload: failed to resolve directory",
zap.String("dir", dir),
zap.Error(err),
)
return
}
if keys, ok := d.dirToKeys[absDir]; ok {
for _, k := range keys {
if k == key {
return
}
}
d.dirToKeys[absDir] = append(keys, key)
return
}
d.dirToKeys[absDir] = []string{key}
watchDirRecursive(d.watcher, absDir, d.logger)
}
func (d *DynamicApp) watchForChanges() {
var debounceTimer *time.Timer
const debounceDuration = 500 * time.Millisecond
pendingDirs := make(map[string]bool)
var pendingMu sync.Mutex
for {
select {
case event, ok := <-d.watcher.Events:
if !ok {
return
}
if !isPythonFileEvent(event) {
handleNewDirEvent(event, d.watcher)
continue
}
d.logger.Debug("python file changed (dynamic)",
zap.String("file", event.Name),
zap.String("op", event.Op.String()),
)
d.mu.RLock()
for absDir := range d.dirToKeys {
if strings.HasPrefix(event.Name, absDir+string(os.PathSeparator)) ||
strings.HasPrefix(event.Name, absDir) {
pendingMu.Lock()
pendingDirs[absDir] = true
pendingMu.Unlock()
}
}
d.mu.RUnlock()
if debounceTimer != nil {
debounceTimer.Stop()
}
debounceTimer = time.AfterFunc(debounceDuration, func() {
pendingMu.Lock()
dirs := make([]string, 0, len(pendingDirs))
for dir := range pendingDirs {
dirs = append(dirs, dir)
}
pendingDirs = make(map[string]bool)
pendingMu.Unlock()
for _, dir := range dirs {
d.reloadDir(dir)
}
})
case err, ok := <-d.watcher.Errors:
if !ok {
return
}
d.logger.Error("autoreload watcher error", zap.Error(err))
case <-d.stopCh:
if debounceTimer != nil {
debounceTimer.Stop()
}
return
}
}
}
// reloadDir evicts all apps associated with the given directory and
// cleans them up after a grace period.
func (d *DynamicApp) reloadDir(absDir string) {
d.logger.Info("reloading dynamic python apps due to file changes",
zap.String("working_dir", absDir),
)
d.mu.Lock()
keys, ok := d.dirToKeys[absDir]
if !ok {
d.mu.Unlock()
return
}
var oldApps []AppServer
for _, key := range keys {
if app, exists := d.apps[key]; exists {
oldApps = append(oldApps, app)
delete(d.apps, key)
}
}
delete(d.dirToKeys, absDir)
d.mu.Unlock()
d.logger.Info("dynamic python apps evicted, will reimport on next request",
zap.String("working_dir", absDir),
zap.Int("apps_evicted", len(oldApps)),
)
if len(oldApps) > 0 {
go func() {
time.Sleep(10 * time.Second)
for _, app := range oldApps {
if err := app.Cleanup(); err != nil {
d.logger.Error("failed to cleanup old dynamic app",
zap.Error(err),
)
}
}
}()
}
}
// HandleRequest resolves placeholders from the request, gets or creates the
// appropriate app, and forwards the request.
func (d *DynamicApp) HandleRequest(w http.ResponseWriter, r *http.Request) error {
key, module, dir, venv, envFiles, envVars := d.resolve(r)
app, err := d.getOrCreateApp(key, module, dir, venv, envFiles, envVars)
if err != nil {
if d.autoreload && d.exitOnReloadFailure != nil {
d.logger.Error("failed to load python app (autoreload); terminating",
zap.String("module", module),
zap.String("working_dir", dir),
zap.Error(err),
)
d.exitOnReloadFailure(1)
}
return err
}
return app.HandleRequest(w, r)
}
// Cleanup frees all dynamically created apps and stops the autoreload watcher.
func (d *DynamicApp) Cleanup() error {
d.mu.Lock()
d.closed = true
for len(d.inflight) > 0 {
waits := make([]chan struct{}, 0, len(d.inflight))
for _, c := range d.inflight {
waits = append(waits, c.done)
}
d.mu.Unlock()
for _, done := range waits {
<-done
}
d.mu.Lock()
}
if d.autoreload && d.stopCh != nil {
close(d.stopCh)
d.watcher.Close()
}
var errs []error
for key, app := range d.apps {
if err := app.Cleanup(); err != nil {
errs = append(errs, err)
}
delete(d.apps, key)
}
d.mu.Unlock()
return errors.Join(errs...)
}
package caddysnake
import (
"bufio"
"fmt"
"os"
"path/filepath"
"regexp"
"strconv"
"strings"
)
var validEnvVarName = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
func validateEnvVarName(name string) error {
if !validEnvVarName.MatchString(name) {
return fmt.Errorf("must match [A-Za-z_][A-Za-z0-9_]*")
}
if name == "PYTHONUNBUFFERED" || strings.HasPrefix(name, "CADDYSNAKE_") {
return fmt.Errorf("reserved environment variable name")
}
return nil
}
func validateEnvVars(envVars map[string]string) error {
for name := range envVars {
if err := validateEnvVarName(name); err != nil {
return fmt.Errorf("invalid env_var name %q: %w", name, err)
}
}
return nil
}
func envMapContainsPlaceholder(envVars map[string]string) bool {
for _, v := range envVars {
if containsPlaceholder(v) {
return true
}
}
return false
}
func envFilesContainPlaceholder(paths []string) bool {
for _, p := range paths {
if containsPlaceholder(p) {
return true
}
}
return false
}
func cloneEnvVars(src map[string]string) map[string]string {
if len(src) == 0 {
return nil
}
dst := make(map[string]string, len(src))
for k, v := range src {
dst[k] = v
}
return dst
}
func cloneEnvFiles(src []string) []string {
if len(src) == 0 {
return nil
}
dst := make([]string, len(src))
copy(dst, src)
return dst
}
func resolveEnvFilePath(workingDir, envFile string) (string, error) {
if envFile == "" {
return "", fmt.Errorf("env_file path is empty")
}
if hasDotDotSegment(envFile) {
return "", fmt.Errorf("env_file path contains path traversal: %q", envFile)
}
path := envFile
if !filepath.IsAbs(path) {
base := workingDir
if base == "" {
var err error
base, err = os.Getwd()
if err != nil {
return "", fmt.Errorf("resolve env_file base directory: %w", err)
}
}
path = filepath.Join(base, path)
}
abs, err := filepath.Abs(path)
if err != nil {
return "", fmt.Errorf("invalid env_file path: %w", err)
}
return abs, nil
}
func parseEnvFile(path string) (map[string]string, error) {
f, err := os.Open(path)
if err != nil {
return nil, err
}
defer f.Close()
vars := make(map[string]string)
scanner := bufio.NewScanner(f)
for lineNum := 1; scanner.Scan(); lineNum++ {
line := strings.TrimSpace(scanner.Text())
if line == "" || strings.HasPrefix(line, "#") {
continue
}
if strings.HasPrefix(line, "export ") {
line = strings.TrimSpace(strings.TrimPrefix(line, "export "))
}
key, value, ok := strings.Cut(line, "=")
if !ok {
return nil, fmt.Errorf("%s:%d: invalid line (expected KEY=VALUE)", path, lineNum)
}
key = strings.TrimSpace(key)
if key == "" {
return nil, fmt.Errorf("%s:%d: empty variable name", path, lineNum)
}
value = strings.TrimSpace(value)
if len(value) >= 2 {
if (value[0] == '"' && value[len(value)-1] == '"') ||
(value[0] == '\'' && value[len(value)-1] == '\'') {
value = value[1 : len(value)-1]
}
}
vars[key] = value
}
if err := scanner.Err(); err != nil {
return nil, err
}
return vars, nil
}
func loadEnvFiles(workingDir string, paths []string) (map[string]string, error) {
merged := make(map[string]string)
for _, p := range paths {
if p == "" {
continue
}
if containsPlaceholder(p) {
return nil, fmt.Errorf("env_file path contains unresolved placeholder: %q", p)
}
abs, err := resolveEnvFilePath(workingDir, p)
if err != nil {
return nil, err
}
info, err := os.Stat(abs)
if err != nil {
return nil, fmt.Errorf("env_file %q: %w", abs, err)
}
if info.IsDir() {
return nil, fmt.Errorf("env_file %q is a directory", abs)
}
vars, err := parseEnvFile(abs)
if err != nil {
return nil, fmt.Errorf("env_file %q: %w", abs, err)
}
for k, v := range vars {
merged[k] = v
}
}
return merged, nil
}
func parseEnvSlice(base []string) map[string]string {
m := make(map[string]string, len(base))
for _, entry := range base {
key, value, ok := strings.Cut(entry, "=")
if !ok {
continue
}
if idx := strings.Index(key, "\x00"); idx >= 0 {
key = key[:idx]
}
m[key] = value
}
return m
}
func envMapToSlice(m map[string]string) []string {
out := make([]string, 0, len(m))
for k, v := range m {
out = append(out, k+"="+v)
}
return out
}
// buildWorkerEnv merges environment variables with precedence:
// base (process env) < fileVars < inlineVars < extra (internal vars).
func buildWorkerEnv(base []string, fileVars, inlineVars map[string]string, extra ...string) []string {
merged := parseEnvSlice(base)
for k, v := range fileVars {
merged[k] = v
}
for k, v := range inlineVars {
merged[k] = v
}
for _, entry := range extra {
key, value, ok := strings.Cut(entry, "=")
if !ok {
continue
}
merged[key] = value
}
return envMapToSlice(merged)
}
func workerInternalEnv(iface, cacheAddr, workerID string) []string {
extra := []string{"PYTHONUNBUFFERED=1"}
if cacheAddr != "" {
extra = append(extra,
EnvCaddysnakeCacheAddr+"="+cacheAddr,
EnvCaddysnakeWorkerInterface+"="+iface,
EnvCaddysnakeCacheTimeoutSeconds+"="+strconv.Itoa(DefaultCacheClientTimeoutSec),
)
if workerID != "" {
extra = append(extra, EnvCaddysnakeWorkerID+"="+workerID)
}
}
return extra
}
//go:build linux
package caddysnake
import (
"os/exec"
"syscall"
)
func setSysProcAttr(cmd *exec.Cmd) {
cmd.SysProcAttr = &syscall.SysProcAttr{
Pdeathsig: syscall.SIGTERM,
}
}
package caddysnake
import (
"context"
"fmt"
"os"
"path/filepath"
"regexp"
"strings"
"github.com/caddyserver/caddy/v2"
"github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
"github.com/caddyserver/caddy/v2/modules/caddytls"
)
func init() {
caddy.RegisterModule(PermissionByPythonDir{})
}
// PermissionByPythonDir implements on-demand TLS permission by checking that the
// requested hostname is of the form {slug}.{domain_suffix} (exactly one label
// before the suffix), and that filepath.Join(root, slug) exists as a directory.
// Users can pair this with a dynamic python block using working_dir "{http.request.host.labels.2}/" when
// slug.appdomain.com uses labels.2 == slug for a three-part host.
//
// Implements [caddytls.OnDemandPermission] as tls.permission.python_dir.
type PermissionByPythonDir struct {
// Absolute base path containing one subdirectory per slug (branch name, tenant, etc.).
Root string `json:"root,omitempty"`
// Registered domain suffix, e.g. appdomain.com (no leading dot). Hostname must be {slug}.{domain_suffix}.
DomainSuffix string `json:"domain_suffix,omitempty"`
rootAbs string
suffixNorm string
}
// CaddyModule returns the Caddy module information.
func (PermissionByPythonDir) CaddyModule() caddy.ModuleInfo {
return caddy.ModuleInfo{
ID: "tls.permission.python_dir",
New: func() caddy.Module { return new(PermissionByPythonDir) },
}
}
// Ungrouped names so a duplicate legacy file can't break the package build alongside this one.
var pythonDirTLSPermSlug = regexp.MustCompile(`^(?:[a-z0-9]|[a-z0-9][a-z0-9_-]*[a-z0-9])$`)
// Provision validates configuration and resolves Root to an absolute path.
func (p *PermissionByPythonDir) Provision(ctx caddy.Context) error {
_ = ctx
if strings.TrimSpace(p.Root) == "" {
return fmt.Errorf("tls.permission.python_dir: root is required")
}
if strings.TrimSpace(p.DomainSuffix) == "" {
return fmt.Errorf("tls.permission.python_dir: domain_suffix is required")
}
abs, err := filepath.Abs(p.Root)
if err != nil {
return fmt.Errorf("tls.permission.python_dir: resolving root: %w", err)
}
rootEval, err := filepath.EvalSymlinks(abs)
if err != nil {
return fmt.Errorf("tls.permission.python_dir: resolving root symlink: %w", err)
}
st, err := os.Stat(rootEval)
if err != nil {
return fmt.Errorf("tls.permission.python_dir: stat root %q: %w", rootEval, err)
}
if !st.IsDir() {
return fmt.Errorf("tls.permission.python_dir: root %q is not a directory", rootEval)
}
p.rootAbs = filepath.Clean(rootEval)
s := strings.ToLower(strings.TrimSpace(p.DomainSuffix))
s = strings.Trim(s, ".")
p.suffixNorm = s
return nil
}
func pythonDirTLSPermInsideRoot(rootCleanAbs, resolved string) bool {
rel, err := filepath.Rel(rootCleanAbs, resolved)
if err != nil {
return false
}
return rel != ".." && !strings.HasPrefix(rel, ".."+string(os.PathSeparator))
}
// CertificateAllowed implements [caddytls.OnDemandPermission].
func (p *PermissionByPythonDir) CertificateAllowed(_ context.Context, name string) error {
host := strings.ToLower(strings.TrimSuffix(strings.TrimSpace(name), "."))
if host == "" {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
suffixNeedle := "." + p.suffixNorm
if !strings.HasSuffix(host, suffixNeedle) {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
slug := strings.TrimSuffix(host, suffixNeedle)
if slug == "" || strings.Contains(slug, ".") {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
if !pythonDirTLSPermSlug.MatchString(slug) {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
dirPath := filepath.Join(p.rootAbs, slug)
dirPath = filepath.Clean(dirPath)
if !pythonDirTLSPermInsideRoot(p.rootAbs, dirPath) {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
dirResolved, err := filepath.EvalSymlinks(dirPath)
if err != nil {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
if !pythonDirTLSPermInsideRoot(p.rootAbs, filepath.Clean(dirResolved)) {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
st, err := os.Stat(dirResolved)
if err != nil || !st.IsDir() {
return fmt.Errorf("%w", caddytls.ErrPermissionDenied)
}
return nil
}
// UnmarshalCaddyfile implements [caddyfile.Unmarshaler].
//
// The Caddy loader prepends the module short name (`python_dir`) before the `{ ... }` block,
// so the first token may be `python_dir` (see caddyfile.UnmarshalModule + NewFromNextSegment).
//
// Example block body:
//
// python_dir {
// root /home/server
// domain_suffix appdomain.com
// }
func (p *PermissionByPythonDir) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
for d.Next() {
switch d.Val() {
case "python_dir":
// optional first token — module name echoed at start of the segment
continue
case "{":
continue
case "}":
continue
case "root":
if !d.Args(&p.Root) {
return d.ArgErr()
}
case "domain_suffix":
if !d.Args(&p.DomainSuffix) {
return d.ArgErr()
}
default:
return d.Errf("unknown subdirective %q", d.Val())
}
}
return nil
}
// Interface guards
var (
_ caddytls.OnDemandPermission = (*PermissionByPythonDir)(nil)
_ caddy.Provisioner = (*PermissionByPythonDir)(nil)
_ caddyfile.Unmarshaler = (*PermissionByPythonDir)(nil)
)