diff --git a/modules/kubernetes/f1-stream/files/.dockerignore b/modules/kubernetes/f1-stream/files/.dockerignore new file mode 100644 index 00000000..ff5e692b --- /dev/null +++ b/modules/kubernetes/f1-stream/files/.dockerignore @@ -0,0 +1,3 @@ +node_modules/ +.claude/ +.git/ diff --git a/modules/kubernetes/f1-stream/files/Dockerfile b/modules/kubernetes/f1-stream/files/Dockerfile index d6625536..6aea6f8d 100644 --- a/modules/kubernetes/f1-stream/files/Dockerfile +++ b/modules/kubernetes/f1-stream/files/Dockerfile @@ -1,4 +1,21 @@ -FROM nginx +FROM golang:1.24-alpine AS builder +WORKDIR /app +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 GOOS=linux go build -o /f1-stream . -COPY . /usr/share/nginx/html -EXPOSE 80 +FROM alpine:3.20 +RUN apk add --no-cache \ + ca-certificates \ + chromium nss freetype harfbuzz ttf-freefont \ + mesa-dri-gallium mesa-gl \ + dbus \ + xvfb-run xorg-server \ + pulseaudio pulseaudio-utils \ + ffmpeg +ENV CHROME_PATH=/usr/bin/chromium-browser +COPY --from=builder /f1-stream /f1-stream +COPY static/ /static/ +EXPOSE 8080 +ENTRYPOINT ["/f1-stream"] diff --git a/modules/kubernetes/f1-stream/files/go.mod b/modules/kubernetes/f1-stream/files/go.mod new file mode 100644 index 00000000..9139f4ab --- /dev/null +++ b/modules/kubernetes/f1-stream/files/go.mod @@ -0,0 +1,45 @@ +module f1-stream + +go 1.24.1 + +require ( + github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327 + github.com/chromedp/chromedp v0.14.2 + github.com/go-webauthn/webauthn v0.15.0 + github.com/gobwas/ws v1.4.0 + github.com/pion/webrtc/v4 v4.2.9 +) + +require ( + github.com/chromedp/sysutil v1.1.0 // indirect + github.com/fxamacker/cbor/v2 v2.9.0 // indirect + github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2 // indirect + github.com/go-viper/mapstructure/v2 v2.4.0 // indirect + github.com/go-webauthn/x v0.1.26 // indirect + github.com/gobwas/httphead v0.1.0 // indirect + github.com/gobwas/pool v0.2.1 // indirect + github.com/golang-jwt/jwt/v5 v5.3.0 // indirect + github.com/google/go-tpm v0.9.6 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/pion/datachannel v1.6.0 // indirect + github.com/pion/dtls/v3 v3.1.2 // indirect + github.com/pion/ice/v4 v4.2.1 // indirect + github.com/pion/interceptor v0.1.44 // indirect + github.com/pion/logging v0.2.4 // indirect + github.com/pion/mdns/v2 v2.1.0 // indirect + github.com/pion/randutil v0.1.0 // indirect + github.com/pion/rtcp v1.2.16 // indirect + github.com/pion/rtp v1.10.1 // indirect + github.com/pion/sctp v1.9.2 // indirect + github.com/pion/sdp/v3 v3.0.18 // indirect + github.com/pion/srtp/v3 v3.0.10 // indirect + github.com/pion/stun/v3 v3.1.1 // indirect + github.com/pion/transport/v4 v4.0.1 // indirect + github.com/pion/turn/v4 v4.1.4 // indirect + github.com/wlynxg/anet v0.0.5 // indirect + github.com/x448/float16 v0.8.4 // indirect + golang.org/x/crypto v0.48.0 // indirect + golang.org/x/net v0.50.0 // indirect + golang.org/x/sys v0.41.0 // indirect + golang.org/x/time v0.10.0 // indirect +) diff --git a/modules/kubernetes/f1-stream/files/go.sum b/modules/kubernetes/f1-stream/files/go.sum new file mode 100644 index 00000000..1ed62f3b --- /dev/null +++ b/modules/kubernetes/f1-stream/files/go.sum @@ -0,0 +1,89 @@ +github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327 h1:UQ4AU+BGti3Sy/aLU8KVseYKNALcX9UXY6DfpwQ6J8E= +github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327/go.mod h1:NItd7aLkcfOA/dcMXvl8p1u+lQqioRMq/SqDp71Pb/k= +github.com/chromedp/chromedp v0.14.2 h1:r3b/WtwM50RsBZHMUm9fsNhhzRStTHrKdr2zmwbZSzM= +github.com/chromedp/chromedp v0.14.2/go.mod h1:rHzAv60xDE7VNy/MYtTUrYreSc0ujt2O1/C3bzctYBo= +github.com/chromedp/sysutil v1.1.0 h1:PUFNv5EcprjqXZD9nJb9b/c9ibAbxiYo4exNWZyipwM= +github.com/chromedp/sysutil v1.1.0/go.mod h1:WiThHUdltqCNKGc4gaU50XgYjwjYIhKWoHGPTUfWTJ8= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= +github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2 h1:iizUGZ9pEquQS5jTGkh4AqeeHCMbfbjeb0zMt0aEFzs= +github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2/go.mod h1:TiCD2a1pcmjd7YnhGH0f/zKNcCD06B029pHhzV23c2M= +github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= +github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= +github.com/go-webauthn/webauthn v0.15.0 h1:LR1vPv62E0/6+sTenX35QrCmpMCzLeVAcnXeH4MrbJY= +github.com/go-webauthn/webauthn v0.15.0/go.mod h1:hcAOhVChPRG7oqG7Xj6XKN1mb+8eXTGP/B7zBLzkX5A= +github.com/go-webauthn/x v0.1.26 h1:eNzreFKnwNLDFoywGh9FA8YOMebBWTUNlNSdolQRebs= +github.com/go-webauthn/x v0.1.26/go.mod h1:jmf/phPV6oIsF6hmdVre+ovHkxjDOmNH0t6fekWUxvg= +github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU= +github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= +github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og= +github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/ws v1.4.0 h1:CTaoG1tojrh4ucGPcoJFiAQUAsEWekEWvLy7GsVNqGs= +github.com/gobwas/ws v1.4.0/go.mod h1:G3gNqMNtPppf5XUz7O4shetPpcZ1VJ7zt18dlUeakrc= +github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo= +github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/google/go-tpm v0.9.6 h1:Ku42PT4LmjDu1H5C5ISWLlpI1mj+Zq7sPGKoRw2XROA= +github.com/google/go-tpm v0.9.6/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80 h1:6Yzfa6GP0rIo/kULo2bwGEkFvCePZ3qHDDTC3/J9Swo= +github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80/go.mod h1:imJHygn/1yfhB7XSJJKlFZKl/J+dCPAknuiaGOshXAs= +github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde h1:x0TT0RDC7UhAVbbWWBzr41ElhJx5tXPWkIHA2HWPRuw= +github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0= +github.com/pion/datachannel v1.6.0 h1:XecBlj+cvsxhAMZWFfFcPyUaDZtd7IJvrXqlXD/53i0= +github.com/pion/datachannel v1.6.0/go.mod h1:ur+wzYF8mWdC+Mkis5Thosk+u/VOL287apDNEbFpsIk= +github.com/pion/dtls/v3 v3.1.2 h1:gqEdOUXLtCGW+afsBLO0LtDD8GnuBBjEy6HRtyofZTc= +github.com/pion/dtls/v3 v3.1.2/go.mod h1:Hw/igcX4pdY69z1Hgv5x7wJFrUkdgHwAn/Q/uo7YHRo= +github.com/pion/ice/v4 v4.2.1 h1:XPRYXaLiFq3LFDG7a7bMrmr3mFr27G/gtXN3v/TVfxY= +github.com/pion/ice/v4 v4.2.1/go.mod h1:2quLV1S5v1tAx3VvAJaH//KGitRXvo4RKlX6D3tnN+c= +github.com/pion/interceptor v0.1.44 h1:sNlZwM8dWXU9JQAkJh8xrarC0Etn8Oolcniukmuy0/I= +github.com/pion/interceptor v0.1.44/go.mod h1:4atVlBkcgXuUP+ykQF0qOCGU2j7pQzX2ofvPRFsY5RY= +github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= +github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so= +github.com/pion/mdns/v2 v2.1.0 h1:3IJ9+Xio6tWYjhN6WwuY142P/1jA0D5ERaIqawg/fOY= +github.com/pion/mdns/v2 v2.1.0/go.mod h1:pcez23GdynwcfRU1977qKU0mDxSeucttSHbCSfFOd9A= +github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= +github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/rtcp v1.2.16 h1:fk1B1dNW4hsI78XUCljZJlC4kZOPk67mNRuQ0fcEkSo= +github.com/pion/rtcp v1.2.16/go.mod h1:/as7VKfYbs5NIb4h6muQ35kQF/J0ZVNz2Z3xKoCBYOo= +github.com/pion/rtp v1.10.1 h1:xP1prZcCTUuhO2c83XtxyOHJteISg6o8iPsE2acaMtA= +github.com/pion/rtp v1.10.1/go.mod h1:rF5nS1GqbR7H/TCpKwylzeq6yDM+MM6k+On5EgeThEM= +github.com/pion/sctp v1.9.2 h1:HxsOzEV9pWoeggv7T5kewVkstFNcGvhMPx0GvUOUQXo= +github.com/pion/sctp v1.9.2/go.mod h1:OTOlsQ5EDQ6mQ0z4MUGXt2CgQmKyafBEXhUVqLRB6G8= +github.com/pion/sdp/v3 v3.0.18 h1:l0bAXazKHpepazVdp+tPYnrsy9dfh7ZbT8DxesH5ZnI= +github.com/pion/sdp/v3 v3.0.18/go.mod h1:ZREGo6A9ZygQ9XkqAj5xYCQtQpif0i6Pa81HOiAdqQ8= +github.com/pion/srtp/v3 v3.0.10 h1:tFirkpBb3XccP5VEXLi50GqXhv5SKPxqrdlhDCJlZrQ= +github.com/pion/srtp/v3 v3.0.10/go.mod h1:3mOTIB0cq9qlbn59V4ozvv9ClW/BSEbRp4cY0VtaR7M= +github.com/pion/stun/v3 v3.1.1 h1:CkQxveJ4xGQjulGSROXbXq94TAWu8gIX2dT+ePhUkqw= +github.com/pion/stun/v3 v3.1.1/go.mod h1:qC1DfmcCTQjl9PBaMa5wSn3x9IPmKxSdcCsxBcDBndM= +github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkYOM= +github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ= +github.com/pion/transport/v4 v4.0.1 h1:sdROELU6BZ63Ab7FrOLn13M6YdJLY20wldXW2Cu2k8o= +github.com/pion/transport/v4 v4.0.1/go.mod h1:nEuEA4AD5lPdcIegQDpVLgNoDGreqM/YqmEx3ovP4jM= +github.com/pion/turn/v4 v4.1.4 h1:EU11yMXKIsK43FhcUnjLlrhE4nboHZq+TXBIi3QpcxQ= +github.com/pion/turn/v4 v4.1.4/go.mod h1:ES1DXVFKnOhuDkqn9hn5VJlSWmZPaRJLyBXoOeO/BmQ= +github.com/pion/webrtc/v4 v4.2.9 h1:DZIh1HAhPIL3RvwEDFsmL5hfPSLEpxsQk9/Jir2vkJE= +github.com/pion/webrtc/v4 v4.2.9/go.mod h1:9EmLZve0H76eTzf8v2FmchZ6tcBXtDgpfTEu+drW6SY= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= +github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= +go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= +golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60= +golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= +golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/time v0.10.0 h1:3usCWA8tQn0L8+hFJQNgzpWbd89begxN66o1Ojdn5L4= +golang.org/x/time v0.10.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/modules/kubernetes/f1-stream/files/internal/auth/auth.go b/modules/kubernetes/f1-stream/files/internal/auth/auth.go new file mode 100644 index 00000000..f73121dc --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/auth/auth.go @@ -0,0 +1,359 @@ +package auth + +import ( + "crypto/rand" + "encoding/json" + "fmt" + "log" + "net/http" + "regexp" + "sync" + "time" + + "f1-stream/internal/models" + "f1-stream/internal/store" + + "github.com/go-webauthn/webauthn/webauthn" +) + +var usernameRe = regexp.MustCompile(`^[a-zA-Z0-9_]{3,30}$`) + +type Auth struct { + store *store.Store + webauthn *webauthn.WebAuthn + adminUsername string + sessionTTL time.Duration + + // In-memory storage for WebAuthn ceremony session data (short-lived) + regSessions map[string]*webauthn.SessionData + loginSessions map[string]*webauthn.SessionData + mu sync.Mutex +} + +func New(s *store.Store, rpDisplayName, rpID string, rpOrigins []string, adminUsername string, sessionTTL time.Duration) (*Auth, error) { + wconfig := &webauthn.Config{ + RPDisplayName: rpDisplayName, + RPID: rpID, + RPOrigins: rpOrigins, + } + w, err := webauthn.New(wconfig) + if err != nil { + return nil, fmt.Errorf("webauthn init: %w", err) + } + return &Auth{ + store: s, + webauthn: w, + adminUsername: adminUsername, + sessionTTL: sessionTTL, + regSessions: make(map[string]*webauthn.SessionData), + loginSessions: make(map[string]*webauthn.SessionData), + }, nil +} + +// BeginRegistration starts the WebAuthn registration ceremony. +func (a *Auth) BeginRegistration(w http.ResponseWriter, r *http.Request) { + var req struct { + Username string `json:"username"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, `{"error":"invalid request"}`, http.StatusBadRequest) + return + } + if !usernameRe.MatchString(req.Username) { + http.Error(w, `{"error":"username must be 3-30 chars, alphanumeric or underscore"}`, http.StatusBadRequest) + return + } + + existing, err := a.store.GetUserByName(req.Username) + if err != nil { + http.Error(w, `{"error":"internal error"}`, http.StatusInternalServerError) + return + } + if existing != nil { + http.Error(w, `{"error":"username already taken"}`, http.StatusConflict) + return + } + + id, err := randomID() + if err != nil { + http.Error(w, `{"error":"internal error"}`, http.StatusInternalServerError) + return + } + + isAdmin := false + if a.adminUsername != "" && req.Username == a.adminUsername { + isAdmin = true + } else if a.adminUsername == "" { + count, err := a.store.UserCount() + if err == nil && count == 0 { + isAdmin = true + } + } + + user := &models.User{ + ID: id, + Username: req.Username, + IsAdmin: isAdmin, + CreatedAt: time.Now(), + } + + options, session, err := a.webauthn.BeginRegistration(user) + if err != nil { + log.Printf("BeginRegistration error: %v", err) + http.Error(w, `{"error":"failed to begin registration"}`, http.StatusInternalServerError) + return + } + + a.mu.Lock() + a.regSessions[req.Username] = session + a.mu.Unlock() + + // Clean up session after 5 minutes + go func() { + time.Sleep(5 * time.Minute) + a.mu.Lock() + delete(a.regSessions, req.Username) + a.mu.Unlock() + }() + + // Store user temporarily - will be committed on finish + // We create the user now so FinishRegistration can look it up + if err := a.store.CreateUser(*user); err != nil { + http.Error(w, `{"error":"failed to create user"}`, http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(options) +} + +// FinishRegistration completes the WebAuthn registration ceremony. +func (a *Auth) FinishRegistration(w http.ResponseWriter, r *http.Request) { + var req struct { + Username string `json:"username"` + } + // Username is passed as query param since body is the attestation response + username := r.URL.Query().Get("username") + if username == "" { + // Try to decode from a wrapper + http.Error(w, `{"error":"username required"}`, http.StatusBadRequest) + return + } + req.Username = username + + a.mu.Lock() + session, ok := a.regSessions[req.Username] + if ok { + delete(a.regSessions, req.Username) + } + a.mu.Unlock() + + if !ok { + http.Error(w, `{"error":"no registration in progress"}`, http.StatusBadRequest) + return + } + + user, err := a.store.GetUserByName(req.Username) + if err != nil || user == nil { + http.Error(w, `{"error":"user not found"}`, http.StatusBadRequest) + return + } + + credential, err := a.webauthn.FinishRegistration(user, *session, r) + if err != nil { + log.Printf("FinishRegistration error: %v", err) + http.Error(w, `{"error":"registration failed"}`, http.StatusBadRequest) + return + } + + user.Credentials = append(user.Credentials, *credential) + if err := a.store.UpdateUserCredentials(user.ID, user.Credentials); err != nil { + http.Error(w, `{"error":"failed to save credential"}`, http.StatusInternalServerError) + return + } + + // Create session + token, err := a.store.CreateSession(user.ID, a.sessionTTL) + if err != nil { + http.Error(w, `{"error":"failed to create session"}`, http.StatusInternalServerError) + return + } + http.SetCookie(w, &http.Cookie{ + Name: "session", + Value: token, + Path: "/", + HttpOnly: true, + SameSite: http.SameSiteStrictMode, + Secure: r.TLS != nil, + MaxAge: int(a.sessionTTL.Seconds()), + }) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "id": user.ID, + "username": user.Username, + "is_admin": user.IsAdmin, + }) +} + +// BeginLogin starts the WebAuthn login ceremony. +func (a *Auth) BeginLogin(w http.ResponseWriter, r *http.Request) { + var req struct { + Username string `json:"username"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, `{"error":"invalid request"}`, http.StatusBadRequest) + return + } + + user, err := a.store.GetUserByName(req.Username) + if err != nil { + http.Error(w, `{"error":"internal error"}`, http.StatusInternalServerError) + return + } + if user == nil { + http.Error(w, `{"error":"user not found"}`, http.StatusNotFound) + return + } + if len(user.Credentials) == 0 { + http.Error(w, `{"error":"no credentials registered"}`, http.StatusBadRequest) + return + } + + options, session, err := a.webauthn.BeginLogin(user) + if err != nil { + log.Printf("BeginLogin error: %v", err) + http.Error(w, `{"error":"failed to begin login"}`, http.StatusInternalServerError) + return + } + + a.mu.Lock() + a.loginSessions[req.Username] = session + a.mu.Unlock() + + go func() { + time.Sleep(5 * time.Minute) + a.mu.Lock() + delete(a.loginSessions, req.Username) + a.mu.Unlock() + }() + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(options) +} + +// FinishLogin completes the WebAuthn login ceremony. +func (a *Auth) FinishLogin(w http.ResponseWriter, r *http.Request) { + username := r.URL.Query().Get("username") + if username == "" { + http.Error(w, `{"error":"username required"}`, http.StatusBadRequest) + return + } + + a.mu.Lock() + session, ok := a.loginSessions[username] + if ok { + delete(a.loginSessions, username) + } + a.mu.Unlock() + + if !ok { + http.Error(w, `{"error":"no login in progress"}`, http.StatusBadRequest) + return + } + + user, err := a.store.GetUserByName(username) + if err != nil || user == nil { + http.Error(w, `{"error":"user not found"}`, http.StatusBadRequest) + return + } + + credential, err := a.webauthn.FinishLogin(user, *session, r) + if err != nil { + log.Printf("FinishLogin error: %v", err) + http.Error(w, `{"error":"login failed"}`, http.StatusUnauthorized) + return + } + + // Update credential sign count + for i, c := range user.Credentials { + if string(c.ID) == string(credential.ID) { + user.Credentials[i].Authenticator.SignCount = credential.Authenticator.SignCount + break + } + } + if err := a.store.UpdateUserCredentials(user.ID, user.Credentials); err != nil { + log.Printf("Failed to update credential sign count: %v", err) + } + + token, err := a.store.CreateSession(user.ID, a.sessionTTL) + if err != nil { + http.Error(w, `{"error":"failed to create session"}`, http.StatusInternalServerError) + return + } + http.SetCookie(w, &http.Cookie{ + Name: "session", + Value: token, + Path: "/", + HttpOnly: true, + SameSite: http.SameSiteStrictMode, + Secure: r.TLS != nil, + MaxAge: int(a.sessionTTL.Seconds()), + }) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "id": user.ID, + "username": user.Username, + "is_admin": user.IsAdmin, + }) +} + +// Logout clears the session. +func (a *Auth) Logout(w http.ResponseWriter, r *http.Request) { + cookie, err := r.Cookie("session") + if err == nil { + a.store.DeleteSession(cookie.Value) + } + http.SetCookie(w, &http.Cookie{ + Name: "session", + Value: "", + Path: "/", + HttpOnly: true, + MaxAge: -1, + }) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"ok":true}`)) +} + +// Me returns the current user info. +func (a *Auth) Me(w http.ResponseWriter, r *http.Request) { + user := UserFromContext(r.Context()) + if user == nil { + http.Error(w, `{"error":"not authenticated"}`, http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "id": user.ID, + "username": user.Username, + "is_admin": user.IsAdmin, + }) +} + +// GetSessionUser returns the user for a session token. +func (a *Auth) GetSessionUser(token string) (*models.User, error) { + sess, err := a.store.GetSession(token) + if err != nil || sess == nil { + return nil, err + } + return a.store.GetUserByID(sess.UserID) +} + +func randomID() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", err + } + return fmt.Sprintf("%x", b), nil +} diff --git a/modules/kubernetes/f1-stream/files/internal/auth/context.go b/modules/kubernetes/f1-stream/files/internal/auth/context.go new file mode 100644 index 00000000..8cc314f4 --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/auth/context.go @@ -0,0 +1,20 @@ +package auth + +import ( + "context" + + "f1-stream/internal/models" +) + +type contextKey string + +const userKey contextKey = "user" + +func ContextWithUser(ctx context.Context, user *models.User) context.Context { + return context.WithValue(ctx, userKey, user) +} + +func UserFromContext(ctx context.Context) *models.User { + user, _ := ctx.Value(userKey).(*models.User) + return user +} diff --git a/modules/kubernetes/f1-stream/files/internal/extractor/browser.go b/modules/kubernetes/f1-stream/files/internal/extractor/browser.go new file mode 100644 index 00000000..740d7eaa --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/extractor/browser.go @@ -0,0 +1,38 @@ +package extractor + +import ( + "log" + "os/exec" +) + +const maxConcurrentSessions = 10 + +var sessionSem chan struct{} + +// Init starts dbus, PulseAudio, and prepares the session semaphore. +func Init() { + // Start dbus (Chrome needs it for accessibility/service queries) + if err := exec.Command("mkdir", "-p", "/var/run/dbus").Run(); err == nil { + if err := exec.Command("dbus-daemon", "--system", "--nofork").Start(); err != nil { + log.Printf("extractor: warning: failed to start dbus: %v", err) + } + } + + if err := exec.Command("pulseaudio", "--start", "--exit-idle-time=-1").Run(); err != nil { + log.Printf("extractor: warning: failed to start PulseAudio: %v", err) + } + // Create a null-sink as the default audio target for all sessions + exec.Command("pactl", "load-module", "module-null-sink", + "sink_name=virtual_sink", + "sink_properties=device.description=VirtualSink").Run() + exec.Command("pactl", "set-default-sink", "virtual_sink").Run() + + sessionSem = make(chan struct{}, maxConcurrentSessions) + log.Println("extractor: initialized") +} + +// Stop kills PulseAudio. +func Stop() { + exec.Command("pulseaudio", "--kill").Run() + log.Println("extractor: stopped") +} diff --git a/modules/kubernetes/f1-stream/files/internal/extractor/capture.go b/modules/kubernetes/f1-stream/files/internal/extractor/capture.go new file mode 100644 index 00000000..b58bcec3 --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/extractor/capture.go @@ -0,0 +1,167 @@ +package extractor + +import ( + "fmt" + "log" + "os" + "os/exec" + "sync/atomic" + "time" +) + +var displayCounter int64 = 99 + +func nextDisplay() int { + return int(atomic.AddInt64(&displayCounter, 1)) +} + +// Capture manages an Xvfb display and separate ffmpeg pipelines for video and audio. +// Audio capture is best-effort — if PulseAudio is unavailable, video still works. +type Capture struct { + display int + xvfbCmd *exec.Cmd + videoCmd *exec.Cmd + audioCmd *exec.Cmd + videoR *os.File // IVF pipe reader (VP8 frames) + audioR *os.File // OGG pipe reader (Opus frames) +} + +// NewCapture starts Xvfb on the given display and two ffmpeg processes: +// one for video (x11grab → VP8/IVF) and one for audio (pulse → Opus/OGG). +// Audio is best-effort — if it fails to start, video still works and audioR +// is set to a pipe that will return EOF immediately. +func NewCapture(display, width, height int) (*Capture, error) { + c := &Capture{display: display} + + // Start Xvfb + screen := fmt.Sprintf("%dx%dx24", width, height) + c.xvfbCmd = exec.Command("Xvfb", fmt.Sprintf(":%d", display), + "-screen", "0", screen, "-ac", "-nolisten", "tcp") + if err := c.xvfbCmd.Start(); err != nil { + return nil, fmt.Errorf("capture: failed to start Xvfb: %w", err) + } + + // Wait for Xvfb to be ready (X11 socket must exist) + ready := false + for i := 0; i < 50; i++ { + socketPath := fmt.Sprintf("/tmp/.X11-unix/X%d", display) + if _, err := os.Stat(socketPath); err == nil { + ready = true + break + } + time.Sleep(100 * time.Millisecond) + } + if !ready { + c.xvfbCmd.Process.Kill() + c.xvfbCmd.Wait() + return nil, fmt.Errorf("capture: Xvfb did not start in time for display :%d", display) + } + + // --- Video pipeline (required) --- + videoR, videoW, err := os.Pipe() + if err != nil { + c.cleanup() + return nil, fmt.Errorf("capture: video pipe: %w", err) + } + + c.videoCmd = exec.Command("ffmpeg", + "-loglevel", "warning", + "-f", "x11grab", "-framerate", "30", + "-video_size", fmt.Sprintf("%dx%d", width, height), + "-i", fmt.Sprintf(":%d", display), + "-c:v", "libvpx", + "-quality", "realtime", "-cpu-used", "8", + "-deadline", "realtime", "-b:v", "2M", "-g", "30", + "-f", "ivf", "pipe:3", + ) + c.videoCmd.ExtraFiles = []*os.File{videoW} + c.videoCmd.Stdout = os.Stderr + c.videoCmd.Stderr = os.Stderr + + if err := c.videoCmd.Start(); err != nil { + videoR.Close() + videoW.Close() + c.cleanup() + return nil, fmt.Errorf("capture: failed to start video ffmpeg: %w", err) + } + videoW.Close() + c.videoR = videoR + + go func() { + if err := c.videoCmd.Wait(); err != nil { + log.Printf("capture: video ffmpeg exited on display :%d: %v", display, err) + } + }() + + // --- Audio pipeline (best-effort) --- + audioR, audioW, err := os.Pipe() + if err != nil { + log.Printf("capture: audio pipe failed on display :%d: %v (continuing without audio)", display, err) + // Provide a closed pipe so StreamAudio gets EOF immediately + r, w, _ := os.Pipe() + w.Close() + c.audioR = r + log.Printf("capture: started display :%d (%dx%d) (video only)", display, width, height) + return c, nil + } + + c.audioCmd = exec.Command("ffmpeg", + "-loglevel", "warning", + "-f", "pulse", "-i", "virtual_sink.monitor", + "-c:a", "libopus", + "-b:a", "128k", "-application", "lowdelay", + "-f", "ogg", "pipe:3", + ) + c.audioCmd.ExtraFiles = []*os.File{audioW} + c.audioCmd.Stdout = os.Stderr + c.audioCmd.Stderr = os.Stderr + + if err := c.audioCmd.Start(); err != nil { + log.Printf("capture: audio ffmpeg failed to start on display :%d: %v (continuing without audio)", display, err) + audioR.Close() + audioW.Close() + // Provide a closed pipe so StreamAudio gets EOF immediately + r, w, _ := os.Pipe() + w.Close() + c.audioR = r + c.audioCmd = nil + log.Printf("capture: started display :%d (%dx%d) (video only)", display, width, height) + return c, nil + } + audioW.Close() + c.audioR = audioR + + go func() { + if err := c.audioCmd.Wait(); err != nil { + log.Printf("capture: audio ffmpeg exited on display :%d: %v", display, err) + } + }() + + log.Printf("capture: started display :%d (%dx%d) (video + audio)", display, width, height) + return c, nil +} + +func (c *Capture) cleanup() { + if c.xvfbCmd != nil && c.xvfbCmd.Process != nil { + c.xvfbCmd.Process.Kill() + c.xvfbCmd.Wait() + } +} + +// Close stops ffmpeg processes, Xvfb, and releases pipe resources. +func (c *Capture) Close() { + if c.videoCmd != nil && c.videoCmd.Process != nil { + c.videoCmd.Process.Kill() + } + if c.audioCmd != nil && c.audioCmd.Process != nil { + c.audioCmd.Process.Kill() + } + if c.videoR != nil { + c.videoR.Close() + } + if c.audioR != nil { + c.audioR.Close() + } + c.cleanup() + log.Printf("capture: stopped display :%d", c.display) +} diff --git a/modules/kubernetes/f1-stream/files/internal/extractor/session.go b/modules/kubernetes/f1-stream/files/internal/extractor/session.go new file mode 100644 index 00000000..3178f4a4 --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/extractor/session.go @@ -0,0 +1,383 @@ +package extractor + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net" + "net/http" + "os" + "os/exec" + "sync" + "time" + + "github.com/chromedp/cdproto/fetch" + "github.com/chromedp/cdproto/input" + "github.com/chromedp/cdproto/network" + "github.com/chromedp/cdproto/page" + "github.com/chromedp/chromedp" + "github.com/gobwas/ws" + "github.com/gobwas/ws/wsutil" + "github.com/pion/webrtc/v4" +) + +const ( + sessionTimeout = 5 * time.Minute + defaultViewportW = 1280 + defaultViewportH = 720 + turnCredentialTTL = 24 * time.Hour +) + +var ( + turnURL string + turnSharedSecret string + turnInternalURL string +) + +// SetTURNConfig sets the TURN server URL, shared secret, and optional internal URL. +// The internal URL is used by pion (server-side) to avoid hairpin NAT issues. +// The public URL is sent to the browser client. +func SetTURNConfig(url, secret, internalURL string) { + turnURL = url + turnSharedSecret = secret + turnInternalURL = internalURL + if turnInternalURL == "" { + turnInternalURL = "turn:coturn.coturn.svc.cluster.local:3478" + } + log.Printf("extractor: TURN configured: public=%s internal=%s", url, turnInternalURL) +} + +var adDomains = []string{ + "doubleclick.net", "googlesyndication.com", "googleadservices.com", + "google-analytics.com", "adnxs.com", "criteo.com", "outbrain.com", + "taboola.com", "amazon-adsystem.com", "popads.net", "popcash.net", + "juicyads.com", "exoclick.com", "trafficjunky.com", "propellerads.com", + "adsterra.com", "hilltopads.net", "revcontent.com", "mgid.com", +} + +type inputMsg struct { + Type string `json:"type"` + X float64 `json:"x"` + Y float64 `json:"y"` + Button int `json:"button"` + DeltaX float64 `json:"deltaX"` + DeltaY float64 `json:"deltaY"` + Key string `json:"key"` + Code string `json:"code"` + Mods int `json:"modifiers"` + Width int `json:"width"` + Height int `json:"height"` + SDP string `json:"sdp"` + Candidate *webrtc.ICECandidateInit `json:"candidate"` +} + +// HandleBrowserSession upgrades to WebSocket and runs a remote browser session +// with WebRTC video/audio streaming and CDP input relay. +func HandleBrowserSession(w http.ResponseWriter, r *http.Request, pageURL string) { + // Check session capacity + select { + case sessionSem <- struct{}{}: + defer func() { <-sessionSem }() + default: + http.Error(w, `{"error":"too many active browser sessions"}`, http.StatusServiceUnavailable) + return + } + + conn, _, _, err := ws.UpgradeHTTP(r, w) + if err != nil { + log.Printf("extractor: session: ws upgrade failed: %v", err) + return + } + defer conn.Close() + + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + + // Allocate display and start capture pipeline + display := nextDisplay() + viewW, viewH := defaultViewportW, defaultViewportH + + cap, err := NewCapture(display, viewW, viewH) + if err != nil { + sendWSError(conn, "failed to start capture: "+err.Error()) + log.Printf("extractor: session: capture error: %v", err) + return + } + defer cap.Close() + + // Start Chrome on the virtual display + opts := append(chromedp.DefaultExecAllocatorOptions[:], + chromedp.Flag("headless", false), + chromedp.Flag("no-sandbox", true), + chromedp.Flag("disable-gpu", true), + chromedp.Flag("disable-software-rasterizer", true), + chromedp.Flag("disable-dev-shm-usage", true), + chromedp.Flag("disable-extensions", true), + chromedp.Flag("disable-background-networking", true), + chromedp.ModifyCmdFunc(func(cmd *exec.Cmd) { + cmd.Env = append(os.Environ(), fmt.Sprintf("DISPLAY=:%d", display)) + }), + chromedp.Flag("autoplay-policy", "no-user-gesture-required"), + chromedp.Flag("window-size", fmt.Sprintf("%d,%d", viewW, viewH)), + chromedp.WSURLReadTimeout(30 * time.Second), + ) + allocCtx, allocCancel := chromedp.NewExecAllocator(ctx, opts...) + defer allocCancel() + + tabCtx, tabCancel := chromedp.NewContext(allocCtx) + defer tabCancel() + + var wsMu sync.Mutex + + // Build ICE servers for pion (server-side) — uses internal TURN URL to avoid hairpin NAT + iceServers := []webrtc.ICEServer{ + {URLs: []string{"stun:stun.l.google.com:19302"}}, + } + var turnCreds *TURNCredentials + if turnURL != "" && turnSharedSecret != "" { + // Server-side: use internal k8s DNS for TURN to bypass NAT + internalCreds := GenerateTURNCredentials(turnInternalURL, turnSharedSecret, turnCredentialTTL) + turnCreds = &internalCreds + iceServers = append(iceServers, webrtc.ICEServer{ + URLs: internalCreds.URLs, + Username: internalCreds.Username, + Credential: internalCreds.Credential, + CredentialType: webrtc.ICECredentialTypePassword, + }) + } + + // Build ad-blocking fetch patterns + adPatterns := make([]*fetch.RequestPattern, 0, len(adDomains)) + for _, domain := range adDomains { + adPatterns = append(adPatterns, &fetch.RequestPattern{ + URLPattern: fmt.Sprintf("*://*.%s/*", domain), + }) + } + + // Set up event listeners before navigation + chromedp.ListenTarget(tabCtx, func(ev interface{}) { + switch e := ev.(type) { + case *fetch.EventRequestPaused: + go chromedp.Run(tabCtx, fetch.FailRequest(e.RequestID, network.ErrorReasonBlockedByClient)) + case *page.EventFrameNavigated: + if e.Frame.ParentID == "" { + go sendURLUpdate(tabCtx, conn, &wsMu, e.Frame.URL) + } + case *page.EventNavigatedWithinDocument: + go sendURLUpdate(tabCtx, conn, &wsMu, e.URL) + } + }) + + // Enable fetch interception (ad blocking) and navigate + if err := chromedp.Run(tabCtx, + fetch.Enable().WithPatterns(adPatterns), + chromedp.Navigate(pageURL), + chromedp.WaitReady("body"), + ); err != nil { + sendWSError(conn, "navigation failed") + log.Printf("extractor: session: navigate error for %s: %v", pageURL, err) + return + } + + // Create WebRTC media stream + mediaStream, err := NewMediaStream(iceServers, func(c *webrtc.ICECandidate) { + data, _ := json.Marshal(map[string]interface{}{ + "type": "ice", + "candidate": c.ToJSON(), + }) + wsMu.Lock() + wsutil.WriteServerMessage(conn, ws.OpText, data) + wsMu.Unlock() + }, cancel) + if err != nil { + sendWSError(conn, "WebRTC setup failed") + log.Printf("extractor: session: webrtc error: %v", err) + return + } + defer mediaStream.Close() + + // Create and send SDP offer + sdp, err := mediaStream.Offer() + if err != nil { + sendWSError(conn, "WebRTC offer failed") + log.Printf("extractor: session: offer error: %v", err) + return + } + + // Send ICE config to client — uses PUBLIC TURN URL (for browser to reach from internet) + clientICE := []map[string]interface{}{ + {"urls": []string{"stun:stun.l.google.com:19302"}}, + } + if turnCreds != nil { + // Client-side: use public IP for TURN (browser connects from internet) + publicCreds := GenerateTURNCredentials(turnURL, turnSharedSecret, turnCredentialTTL) + clientICE = append(clientICE, map[string]interface{}{ + "urls": publicCreds.URLs, + "username": publicCreds.Username, + "credential": publicCreds.Credential, + }) + } + iceMsg, _ := json.Marshal(map[string]interface{}{ + "type": "iceServers", + "iceServers": clientICE, + }) + wsMu.Lock() + wsutil.WriteServerMessage(conn, ws.OpText, iceMsg) + wsMu.Unlock() + + offerMsg, _ := json.Marshal(map[string]interface{}{ + "type": "offer", + "sdp": sdp, + }) + wsMu.Lock() + wsutil.WriteServerMessage(conn, ws.OpText, offerMsg) + wsMu.Unlock() + + // Send ready message with viewport dimensions + readyMsg, _ := json.Marshal(map[string]interface{}{ + "type": "ready", + "width": viewW, + "height": viewH, + }) + wsMu.Lock() + wsutil.WriteServerMessage(conn, ws.OpText, readyMsg) + wsMu.Unlock() + + // Start streaming video and audio from capture pipes + go mediaStream.StreamVideo(cap.videoR, ctx) + go mediaStream.StreamAudio(cap.audioR, ctx) + + log.Printf("extractor: session: started for %s (display :%d)", pageURL, display) + + // Inactivity timer — cancels session after no client input + inactivity := time.NewTimer(sessionTimeout) + defer inactivity.Stop() + go func() { + select { + case <-inactivity.C: + log.Printf("extractor: session: inactivity timeout for %s", pageURL) + cancel() + case <-ctx.Done(): + } + }() + + // Read loop — process signaling and input messages + for { + msgs, err := wsutil.ReadClientMessage(conn, nil) + if err != nil { + break + } + for _, m := range msgs { + if m.OpCode != ws.OpText { + continue + } + + // Reset inactivity timer + if !inactivity.Stop() { + select { + case <-inactivity.C: + default: + } + } + inactivity.Reset(sessionTimeout) + + var msg inputMsg + if err := json.Unmarshal(m.Payload, &msg); err != nil { + continue + } + + switch msg.Type { + case "answer": + if err := mediaStream.SetAnswer(msg.SDP); err != nil { + log.Printf("extractor: session: set answer error: %v", err) + } + case "ice": + if msg.Candidate != nil { + if err := mediaStream.AddICECandidate(*msg.Candidate); err != nil { + log.Printf("extractor: session: add ICE error: %v", err) + } + } + case "back": + chromedp.Run(tabCtx, chromedp.NavigateBack()) + case "forward": + chromedp.Run(tabCtx, chromedp.NavigateForward()) + default: + handleInput(tabCtx, &msg) + } + } + } + + log.Printf("extractor: session: ended for %s", pageURL) +} + +func handleInput(ctx context.Context, msg *inputMsg) { + switch msg.Type { + case "mousemove": + chromedp.Run(ctx, + input.DispatchMouseEvent(input.MouseMoved, msg.X, msg.Y)) + case "mousedown": + chromedp.Run(ctx, + input.DispatchMouseEvent(input.MousePressed, msg.X, msg.Y). + WithButton(mapButton(msg.Button)).WithClickCount(1)) + case "mouseup": + chromedp.Run(ctx, + input.DispatchMouseEvent(input.MouseReleased, msg.X, msg.Y). + WithButton(mapButton(msg.Button))) + case "scroll": + chromedp.Run(ctx, + input.DispatchMouseEvent(input.MouseWheel, msg.X, msg.Y). + WithDeltaX(msg.DeltaX).WithDeltaY(msg.DeltaY)) + case "keydown": + chromedp.Run(ctx, + input.DispatchKeyEvent(input.KeyDown). + WithKey(msg.Key).WithCode(msg.Code). + WithModifiers(input.Modifier(msg.Mods))) + case "keyup": + chromedp.Run(ctx, + input.DispatchKeyEvent(input.KeyUp). + WithKey(msg.Key).WithCode(msg.Code). + WithModifiers(input.Modifier(msg.Mods))) + } +} + +func mapButton(jsButton int) input.MouseButton { + switch jsButton { + case 1: + return input.Middle + case 2: + return input.Right + default: + return input.Left + } +} + +func sendURLUpdate(tabCtx context.Context, conn net.Conn, mu *sync.Mutex, currentURL string) { + var canBack, canForward bool + var entries []*page.NavigationEntry + var currentIndex int64 + + if err := chromedp.Run(tabCtx, chromedp.ActionFunc(func(ctx context.Context) error { + var err error + currentIndex, entries, err = page.GetNavigationHistory().Do(ctx) + return err + })); err == nil { + canBack = currentIndex > 0 + canForward = int(currentIndex) < len(entries)-1 + } + + data, _ := json.Marshal(map[string]interface{}{ + "type": "url", + "url": currentURL, + "canBack": canBack, + "canForward": canForward, + }) + mu.Lock() + wsutil.WriteServerMessage(conn, ws.OpText, data) + mu.Unlock() +} + +func sendWSError(conn net.Conn, msg string) { + data, _ := json.Marshal(map[string]string{"type": "error", "message": msg}) + wsutil.WriteServerMessage(conn, ws.OpText, data) +} diff --git a/modules/kubernetes/f1-stream/files/internal/extractor/webrtc.go b/modules/kubernetes/f1-stream/files/internal/extractor/webrtc.go new file mode 100644 index 00000000..e6e1731f --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/extractor/webrtc.go @@ -0,0 +1,248 @@ +package extractor + +import ( + "context" + "crypto/hmac" + "crypto/sha1" + "encoding/base64" + "fmt" + "io" + "log" + "time" + + "github.com/pion/webrtc/v4" + "github.com/pion/webrtc/v4/pkg/media" + "github.com/pion/webrtc/v4/pkg/media/ivfreader" + "github.com/pion/webrtc/v4/pkg/media/oggreader" +) + +// TURNCredentials holds ephemeral TURN credentials generated from a shared secret. +type TURNCredentials struct { + URLs []string `json:"urls"` + Username string `json:"username"` + Credential string `json:"credential"` +} + +// GenerateTURNCredentials creates time-limited TURN credentials using the +// shared secret (TURN REST API / coturn --use-auth-secret). +func GenerateTURNCredentials(turnURL, sharedSecret string, ttl time.Duration) TURNCredentials { + expiry := time.Now().Add(ttl).Unix() + username := fmt.Sprintf("%d", expiry) + + mac := hmac.New(sha1.New, []byte(sharedSecret)) + mac.Write([]byte(username)) + credential := base64.StdEncoding.EncodeToString(mac.Sum(nil)) + + return TURNCredentials{ + URLs: []string{turnURL}, + Username: username, + Credential: credential, + } +} + +// MediaStream wraps a pion WebRTC PeerConnection with VP8 video and Opus audio tracks. +type MediaStream struct { + pc *webrtc.PeerConnection + videoTrack *webrtc.TrackLocalStaticSample + audioTrack *webrtc.TrackLocalStaticSample +} + +// NewMediaStream creates a PeerConnection with VP8 + Opus tracks and an ICE callback. +// The cancel function is called when ICE fails to trigger session cleanup. +func NewMediaStream(iceServers []webrtc.ICEServer, onICE func(*webrtc.ICECandidate), cancel context.CancelFunc) (*MediaStream, error) { + config := webrtc.Configuration{ + ICEServers: iceServers, + } + + pc, err := webrtc.NewPeerConnection(config) + if err != nil { + return nil, err + } + + videoTrack, err := webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, + "video", "stream", + ) + if err != nil { + pc.Close() + return nil, err + } + + audioTrack, err := webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, + "audio", "stream", + ) + if err != nil { + pc.Close() + return nil, err + } + + if _, err = pc.AddTrack(videoTrack); err != nil { + pc.Close() + return nil, err + } + if _, err = pc.AddTrack(audioTrack); err != nil { + pc.Close() + return nil, err + } + + pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) { + log.Printf("webrtc: ICE connection state: %s", state.String()) + if state == webrtc.ICEConnectionStateFailed { + log.Printf("webrtc: ICE failed, cancelling session") + cancel() + return + } + if state == webrtc.ICEConnectionStateConnected { + // Log selected candidate pair + if stats := pc.GetStats(); stats != nil { + for _, s := range stats { + if cp, ok := s.(webrtc.ICECandidatePairStats); ok && cp.Nominated { + log.Printf("webrtc: selected candidate pair: local=%s remote=%s", + cp.LocalCandidateID, cp.RemoteCandidateID) + } + } + } + // Start periodic stats logging + go func() { + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for range ticker.C { + if pc.ICEConnectionState() != webrtc.ICEConnectionStateConnected && + pc.ICEConnectionState() != webrtc.ICEConnectionStateCompleted { + return + } + stats := pc.GetStats() + for _, s := range stats { + if out, ok := s.(webrtc.OutboundRTPStreamStats); ok { + log.Printf("webrtc: outbound-rtp kind=%s bytes=%d packets=%d", + out.Kind, out.BytesSent, out.PacketsSent) + } + } + } + }() + } + }) + + pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { + log.Printf("webrtc: peer connection state: %s", state.String()) + }) + + pc.OnICECandidate(func(c *webrtc.ICECandidate) { + if c != nil { + log.Printf("webrtc: gathered ICE candidate: type=%s addr=%s:%d", + c.Typ.String(), c.Address, c.Port) + if onICE != nil { + onICE(c) + } + } + }) + + return &MediaStream{ + pc: pc, + videoTrack: videoTrack, + audioTrack: audioTrack, + }, nil +} + +// Offer creates an SDP offer, sets it as local description, and returns the SDP string. +func (m *MediaStream) Offer() (string, error) { + offer, err := m.pc.CreateOffer(nil) + if err != nil { + return "", err + } + if err := m.pc.SetLocalDescription(offer); err != nil { + return "", err + } + return offer.SDP, nil +} + +// SetAnswer sets the remote SDP answer. +func (m *MediaStream) SetAnswer(sdp string) error { + return m.pc.SetRemoteDescription(webrtc.SessionDescription{ + Type: webrtc.SDPTypeAnswer, + SDP: sdp, + }) +} + +// AddICECandidate adds a remote ICE candidate. +func (m *MediaStream) AddICECandidate(init webrtc.ICECandidateInit) error { + return m.pc.AddICECandidate(init) +} + +// StreamVideo reads VP8 frames from an IVF stream and writes them to the video track. +// Blocks until the reader returns an error or the context is cancelled. +func (m *MediaStream) StreamVideo(r io.Reader, ctx context.Context) { + ivf, _, err := ivfreader.NewWith(r) + if err != nil { + log.Printf("webrtc: ivf reader error: %v", err) + return + } + + duration := time.Second / 30 + + for { + select { + case <-ctx.Done(): + return + default: + } + + frame, _, err := ivf.ParseNextFrame() + if err != nil { + if err != io.EOF { + log.Printf("webrtc: video frame error: %v", err) + } + return + } + + if err := m.videoTrack.WriteSample(media.Sample{ + Data: frame, + Duration: duration, + }); err != nil { + log.Printf("webrtc: video write error: %v", err) + return + } + } +} + +// StreamAudio reads Opus pages from an OGG stream and writes them to the audio track. +// Blocks until the reader returns an error or the context is cancelled. +func (m *MediaStream) StreamAudio(r io.Reader, ctx context.Context) { + ogg, _, err := oggreader.NewWith(r) + if err != nil { + log.Printf("webrtc: ogg reader error: %v", err) + return + } + + for { + select { + case <-ctx.Done(): + return + default: + } + + page, _, err := ogg.ParseNextPage() + if err != nil { + if err != io.EOF { + log.Printf("webrtc: audio page error: %v", err) + } + return + } + + if err := m.audioTrack.WriteSample(media.Sample{ + Data: page, + Duration: 20 * time.Millisecond, + }); err != nil { + log.Printf("webrtc: audio write error: %v", err) + return + } + } +} + +// Close closes the underlying PeerConnection. +func (m *MediaStream) Close() { + if m.pc != nil { + m.pc.Close() + } +} diff --git a/modules/kubernetes/f1-stream/files/internal/healthcheck/healthcheck.go b/modules/kubernetes/f1-stream/files/internal/healthcheck/healthcheck.go new file mode 100644 index 00000000..217f4803 --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/healthcheck/healthcheck.go @@ -0,0 +1,188 @@ +package healthcheck + +import ( + "context" + "log" + "net/http" + "sync" + "time" + + "f1-stream/internal/models" + "f1-stream/internal/store" +) + +const unhealthyThreshold = 5 + +const userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + +// isReachable sends a GET request and returns true if the server responds with +// an HTTP 2xx or 3xx status code. +func isReachable(client *http.Client, rawURL string) bool { + req, err := http.NewRequest("GET", rawURL, nil) + if err != nil { + return false + } + req.Header.Set("User-Agent", userAgent) + + resp, err := client.Do(req) + if err != nil { + return false + } + defer resp.Body.Close() + + return resp.StatusCode >= 200 && resp.StatusCode < 400 +} + +type HealthChecker struct { + store *store.Store + interval time.Duration + timeout time.Duration + client *http.Client + mu sync.Mutex +} + +func New(s *store.Store, interval, timeout time.Duration) *HealthChecker { + return &HealthChecker{ + store: s, + interval: interval, + timeout: timeout, + client: &http.Client{ + Timeout: timeout, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 3 { + return http.ErrUseLastResponse + } + return nil + }, + }, + } +} + +func (hc *HealthChecker) Run(ctx context.Context) { + log.Printf("healthcheck: starting with interval=%v timeout=%v", hc.interval, hc.timeout) + hc.checkAll() + + ticker := time.NewTicker(hc.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + log.Println("healthcheck: shutting down") + return + case <-ticker.C: + hc.checkAll() + } + } +} + +func (hc *HealthChecker) checkAll() { + hc.mu.Lock() + defer hc.mu.Unlock() + + start := time.Now() + urls := hc.collectURLs() + log.Printf("healthcheck: checking %d URLs", len(urls)) + + existing, err := hc.store.LoadHealthStates() + if err != nil { + log.Printf("healthcheck: failed to load health states: %v", err) + existing = nil + } + + stateMap := make(map[string]*models.HealthState, len(existing)) + for i := range existing { + stateMap[existing[i].URL] = &existing[i] + } + + now := time.Now() + var recovered, newlyUnhealthy int + + for _, url := range urls { + st, exists := stateMap[url] + if !exists { + st = &models.HealthState{ + URL: url, + Healthy: true, + } + stateMap[url] = st + } + + ok := isReachable(hc.client, url) + + if ok { + if !st.Healthy { + log.Printf("healthcheck: recovered %s", truncate(url, 80)) + recovered++ + } + st.ConsecutiveFailures = 0 + st.Healthy = true + } else { + st.ConsecutiveFailures++ + if st.ConsecutiveFailures >= unhealthyThreshold && st.Healthy { + st.Healthy = false + log.Printf("healthcheck: marking unhealthy after %d failures: %s", st.ConsecutiveFailures, truncate(url, 80)) + newlyUnhealthy++ + } + } + st.LastCheckTime = now + } + + // Prune orphaned entries: only keep states whose URL is in the current set + urlSet := make(map[string]bool, len(urls)) + for _, u := range urls { + urlSet[u] = true + } + var finalStates []models.HealthState + healthyCount := 0 + for _, st := range stateMap { + if urlSet[st.URL] { + finalStates = append(finalStates, *st) + if st.Healthy { + healthyCount++ + } + } + } + + if err := hc.store.SaveHealthStates(finalStates); err != nil { + log.Printf("healthcheck: failed to save health states: %v", err) + } + + log.Printf("healthcheck: done in %v, checked=%d healthy=%d recovered=%d newly_unhealthy=%d", + time.Since(start).Round(time.Millisecond), len(urls), healthyCount, recovered, newlyUnhealthy) +} + +func (hc *HealthChecker) collectURLs() []string { + seen := make(map[string]bool) + + streams, err := hc.store.LoadStreams() + if err != nil { + log.Printf("healthcheck: failed to load streams: %v", err) + } else { + for _, s := range streams { + seen[s.URL] = true + } + } + + scraped, err := hc.store.LoadScrapedLinks() + if err != nil { + log.Printf("healthcheck: failed to load scraped links: %v", err) + } else { + for _, l := range scraped { + seen[l.URL] = true + } + } + + urls := make([]string, 0, len(seen)) + for u := range seen { + urls = append(urls, u) + } + return urls +} + +func truncate(s string, maxLen int) string { + if len(s) <= maxLen { + return s + } + return s[:maxLen] + "..." +} diff --git a/modules/kubernetes/f1-stream/files/internal/models/models.go b/modules/kubernetes/f1-stream/files/internal/models/models.go new file mode 100644 index 00000000..b6798cc5 --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/models/models.go @@ -0,0 +1,53 @@ +package models + +import ( + "time" + + "github.com/go-webauthn/webauthn/webauthn" +) + +type User struct { + ID string `json:"id"` + Username string `json:"username"` + IsAdmin bool `json:"is_admin"` + Credentials []webauthn.Credential `json:"credentials"` + CreatedAt time.Time `json:"created_at"` +} + +// WebAuthn interface implementation +func (u *User) WebAuthnID() []byte { return []byte(u.ID) } +func (u *User) WebAuthnName() string { return u.Username } +func (u *User) WebAuthnDisplayName() string { return u.Username } +func (u *User) WebAuthnCredentials() []webauthn.Credential { return u.Credentials } + +type Stream struct { + ID string `json:"id"` + URL string `json:"url"` + Title string `json:"title"` + SubmittedBy string `json:"submitted_by"` + Published bool `json:"published"` + Source string `json:"source"` + CreatedAt time.Time `json:"created_at"` +} + +type ScrapedLink struct { + ID string `json:"id"` + URL string `json:"url"` + Title string `json:"title"` + Source string `json:"source"` + ScrapedAt time.Time `json:"scraped_at"` + Stale bool `json:"stale"` +} + +type Session struct { + Token string `json:"token"` + UserID string `json:"user_id"` + ExpiresAt time.Time `json:"expires_at"` +} + +type HealthState struct { + URL string `json:"url"` + ConsecutiveFailures int `json:"consecutive_failures"` + LastCheckTime time.Time `json:"last_check_time"` + Healthy bool `json:"healthy"` +} diff --git a/modules/kubernetes/f1-stream/files/internal/proxy/proxy.go b/modules/kubernetes/f1-stream/files/internal/proxy/proxy.go new file mode 100644 index 00000000..d631b22f --- /dev/null +++ b/modules/kubernetes/f1-stream/files/internal/proxy/proxy.go @@ -0,0 +1,429 @@ +package proxy + +import ( + "encoding/base64" + "fmt" + "io" + "log" + "net/http" + "net/url" + "regexp" + "strings" +) + +// hopHeaders are headers that should not be forwarded by proxies. +var hopHeaders = map[string]bool{ + "Connection": true, + "Keep-Alive": true, + "Proxy-Authenticate": true, + "Proxy-Authorization": true, + "Te": true, + "Trailers": true, + "Transfer-Encoding": true, + "Upgrade": true, +} + +// antiFrameHeaders are headers we strip to allow iframe embedding. +var antiFrameHeaders = []string{ + "X-Frame-Options", + "Content-Security-Policy", + "Content-Security-Policy-Report-Only", + "X-Content-Type-Options", +} + +// forwardHeaders are request headers we copy from the client to the upstream. +// NOTE: Accept-Encoding is intentionally omitted so Go's Transport handles +// compression transparently (adds gzip, auto-decompresses response body). +// This ensures we can do text replacements on HTML/CSS bodies. +var forwardHeaders = []string{ + "User-Agent", + "Accept", + "Accept-Language", + "Cookie", + "Referer", + "Range", + "If-None-Match", + "If-Modified-Since", + "Cache-Control", +} + +// jsShimTemplate is injected into HTML responses to intercept JS-initiated requests. +// It patches fetch, XMLHttpRequest, WebSocket, and EventSource to route through the proxy. +const jsShimTemplate = `` + +// NewHandler returns an http.Handler that serves the reverse proxy at /proxy/. +// URL structure: /proxy/{base64_origin}/{path...} +func NewHandler() http.Handler { + client := &http.Client{ + Timeout: 30 * 1000000000, // 30s + CheckRedirect: func(req *http.Request, via []*http.Request) error { + return http.ErrUseLastResponse // don't follow redirects + }, + } + + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Parse: /proxy/{base64_origin}/{path...} + trimmed := strings.TrimPrefix(r.URL.Path, "/proxy/") + if trimmed == "" || trimmed == r.URL.Path { + http.Error(w, "bad proxy URL", http.StatusBadRequest) + return + } + + // Split into base64 segment and remaining path + slashIdx := strings.Index(trimmed, "/") + var b64Origin, pathAndQuery string + if slashIdx == -1 { + b64Origin = trimmed + pathAndQuery = "/" + } else { + b64Origin = trimmed[:slashIdx] + pathAndQuery = trimmed[slashIdx:] + } + + originBytes, err := base64.RawURLEncoding.DecodeString(b64Origin) + if err != nil { + // Try standard encoding with padding + originBytes, err = base64.StdEncoding.DecodeString(b64Origin) + if err != nil { + http.Error(w, "invalid base64 origin", http.StatusBadRequest) + return + } + } + origin := string(originBytes) + + // Validate origin is a valid URL + originURL, err := url.Parse(origin) + if err != nil || (originURL.Scheme != "http" && originURL.Scheme != "https") { + http.Error(w, "invalid origin URL", http.StatusBadRequest) + return + } + + // Build upstream URL + targetURL := origin + pathAndQuery + if r.URL.RawQuery != "" { + targetURL += "?" + r.URL.RawQuery + } + + log.Printf("proxy: %s %s -> %s", r.Method, r.URL.Path, targetURL) + + // Create upstream request + upReq, err := http.NewRequestWithContext(r.Context(), r.Method, targetURL, r.Body) + if err != nil { + http.Error(w, "failed to create request", http.StatusInternalServerError) + return + } + + // Copy selected headers + for _, h := range forwardHeaders { + if v := r.Header.Get(h); v != "" { + upReq.Header.Set(h, v) + } + } + // Reconstruct the original Referer from the client's proxy-rewritten Referer. + // The client sends e.g. "https://f1.viktorbarzin.me/proxy/{b64origin}/path" + // and we need to decode that back to "https://original.com/path". + upReq.Header.Set("Referer", decodeProxyReferer(r.Header.Get("Referer"), origin)) + + // Fetch upstream + resp, err := client.Do(upReq) + if err != nil { + log.Printf("proxy: upstream fetch failed: %v", err) + http.Error(w, "upstream fetch failed", http.StatusBadGateway) + return + } + defer resp.Body.Close() + + log.Printf("proxy: %s %s <- %d (%s)", r.Method, r.URL.Path, resp.StatusCode, resp.Header.Get("Content-Type")) + + // Handle redirects: rewrite Location header through proxy + if resp.StatusCode >= 300 && resp.StatusCode < 400 { + loc := resp.Header.Get("Location") + if loc != "" { + rewritten := rewriteRedirect(loc, origin, b64Origin) + w.Header().Set("Location", rewritten) + log.Printf("proxy: redirect %s -> %s", loc, rewritten) + } + w.WriteHeader(resp.StatusCode) + return + } + + // Copy response headers, stripping anti-frame, hop-by-hop, and encoding headers. + // Content-Encoding is stripped because Go's Transport already decompressed the body. + // Content-Length is stripped because we may rewrite the body (changing its length). + for key, vals := range resp.Header { + if hopHeaders[key] { + continue + } + if strings.EqualFold(key, "Content-Encoding") || strings.EqualFold(key, "Content-Length") { + continue + } + skip := false + for _, ah := range antiFrameHeaders { + if strings.EqualFold(key, ah) { + skip = true + break + } + } + if skip { + continue + } + for _, v := range vals { + w.Header().Add(key, v) + } + } + + // Add permissive CORS headers so cross-origin XHR/fetch from the iframe works + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "*") + + w.WriteHeader(resp.StatusCode) + + // For HTML responses, rewrite URLs and inject JS shim + ct := resp.Header.Get("Content-Type") + if strings.Contains(ct, "text/html") { + body, err := io.ReadAll(resp.Body) + if err != nil { + log.Printf("proxy: failed to read HTML body: %v", err) + return + } + rewritten := rewriteHTML(string(body), origin, b64Origin) + w.Write([]byte(rewritten)) + return + } + + // For CSS responses, rewrite url() references + if strings.Contains(ct, "text/css") { + body, err := io.ReadAll(resp.Body) + if err != nil { + log.Printf("proxy: failed to read CSS body: %v", err) + return + } + rewritten := rewriteCSS(string(body), origin, b64Origin) + w.Write([]byte(rewritten)) + return + } + + // Stream other responses directly + io.Copy(w, resp.Body) + }) +} + +// rewriteRedirect rewrites a Location header value to route through the proxy. +func rewriteRedirect(loc, origin, b64Origin string) string { + // Absolute URL on the same origin + if strings.HasPrefix(loc, origin) { + path := strings.TrimPrefix(loc, origin) + return "/proxy/" + b64Origin + path + } + // Absolute URL on a different origin — proxy it too + parsed, err := url.Parse(loc) + if err != nil { + return loc + } + if parsed.IsAbs() { + newOrigin := parsed.Scheme + "://" + parsed.Host + newB64 := base64.RawURLEncoding.EncodeToString([]byte(newOrigin)) + return "/proxy/" + newB64 + parsed.RequestURI() + } + // Relative URL — it will resolve naturally + return loc +} + +// Precompiled regexes for root-relative URL rewriting in HTML attributes. +// Matches src="/...", href="/...", action="/...", poster="/..." but NOT "//..." (protocol-relative). +var rootRelativeAttrRe = regexp.MustCompile(`((?:src|href|action|poster|data)\s*=\s*["'])/([^/"'][^"']*)`) + +// Matches url("/...") or url('/...') or url(/...) in inline styles — but NOT url("//...") +var rootRelativeCSSRe = regexp.MustCompile(`(url\(\s*["']?)/([^/"')[^"')]*)(["']?\s*\))`) + +// crossOriginIframeSrcRe matches