2fa8f2435f
Capture, decrypt, and restream toolkit with compiled-in app modules (RTE, TG4, BBC), on-device MITM proxy, streamd control plane, and www. BBC module.yaml is published (clear streams); other module values stay local.
259 lines
6.5 KiB
Go
259 lines
6.5 KiB
Go
// Package agentcmd is the agent CLI, exposed as a library so the single drm
|
|
// binary can host it as a subcommand.
|
|
package agentcmd
|
|
|
|
import (
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"syscall"
|
|
"text/tabwriter"
|
|
|
|
"drmdecryption/apps/agent/internal/client"
|
|
"drmdecryption/apps/agent/internal/config"
|
|
"drmdecryption/apps/agent/internal/device"
|
|
"drmdecryption/apps/agent/internal/poller"
|
|
"drmdecryption/apps/agent/internal/queue"
|
|
"drmdecryption/apps/agent/internal/worker"
|
|
"drmdecryption/apps/agent/internal/wsclient"
|
|
)
|
|
|
|
// Run dispatches an agent subcommand. args[0] is the subcommand name.
|
|
func Run(args []string) error {
|
|
log.SetFlags(log.LstdFlags | log.Lmsgprefix)
|
|
log.SetPrefix("agent: ")
|
|
|
|
if len(args) < 1 {
|
|
Usage()
|
|
return fmt.Errorf("agent: subcommand required")
|
|
}
|
|
sub, rest := args[0], args[1:]
|
|
switch sub {
|
|
case "run":
|
|
runCmd(sub, rest)
|
|
case "status":
|
|
statusCmd(sub, rest)
|
|
case "devices":
|
|
devicesCmd(sub, rest)
|
|
case "enqueue":
|
|
enqueueCmd(rest)
|
|
case "cancel":
|
|
cancelCmd(rest)
|
|
case "help", "-h", "--help":
|
|
Usage()
|
|
default:
|
|
Usage()
|
|
return fmt.Errorf("agent: unknown subcommand %q", sub)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Usage prints the agent subcommand help.
|
|
func Usage() {
|
|
fmt.Fprintf(os.Stderr, `agent — always-on credentials refresher (queue + ADB)
|
|
|
|
Usage:
|
|
agent run [--config apps/agent/agent.yaml]
|
|
agent status [--config ...]
|
|
agent devices [--config ...]
|
|
agent enqueue --stream NAME [--reason manual] [--config ...]
|
|
agent cancel --job ID [--config ...]
|
|
|
|
Polls streamd for enabled streams that are down / missing credentials,
|
|
enqueues jobs, claims a free phone that already has the app, captures
|
|
via MITM, POSTs credentials, then force-stops the app.
|
|
|
|
Does not install apps. Does not run NRE/ffmpeg.
|
|
`)
|
|
}
|
|
|
|
func loadCfg(sub string, args []string) (config.Config, *flag.FlagSet) {
|
|
fs := flag.NewFlagSet(sub, flag.ExitOnError)
|
|
cfgPath := fs.String("config", "", "path to apps/agent/agent.yaml")
|
|
_ = fs.Parse(args)
|
|
cfg, err := config.Load(*cfgPath)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
return cfg, fs
|
|
}
|
|
|
|
func openQueue(cfg config.Config) *queue.Store {
|
|
q, err := queue.Open(cfg.DataDir)
|
|
if err != nil {
|
|
log.Fatalf("queue: %v", err)
|
|
}
|
|
return q
|
|
}
|
|
|
|
func runCmd(sub string, args []string) {
|
|
cfg, _ := loadCfg(sub, args)
|
|
api := client.New(cfg.StreamdURL, cfg.Token)
|
|
if err := api.Health(); err != nil {
|
|
log.Fatalf("streamd unreachable at %s: %v", cfg.StreamdURL, err)
|
|
}
|
|
q := openQueue(cfg)
|
|
defer q.Close()
|
|
|
|
if n, err := q.RecoverStaleRunning(); err != nil {
|
|
log.Fatalf("recover jobs: %v", err)
|
|
} else if n > 0 {
|
|
log.Printf("re-queued %d stuck running job(s) from previous run", n)
|
|
}
|
|
|
|
pool := device.New("", cfg.Devices)
|
|
stop := make(chan struct{})
|
|
|
|
p := &poller.Poller{
|
|
API: api,
|
|
Queue: q,
|
|
MaxAttempts: cfg.MaxAttempts,
|
|
Interval: cfg.PollInterval(),
|
|
Log: log.Default(),
|
|
}
|
|
go p.Loop(stop)
|
|
|
|
ws := wsclient.New(cfg.StreamdURL, cfg.Token, cfg.AgentID, log.Default())
|
|
go ws.Loop()
|
|
|
|
for i := 0; i < cfg.Workers; i++ {
|
|
w := &worker.Worker{
|
|
ID: i + 1,
|
|
API: api,
|
|
Queue: q,
|
|
Pool: pool,
|
|
AgentID: cfg.AgentID,
|
|
CaptureWait: cfg.CaptureWait(),
|
|
Python: cfg.Python,
|
|
WVD: cfg.WVD,
|
|
UserAgent: cfg.UserAgent,
|
|
Log: log.Default(),
|
|
}
|
|
go w.Loop(stop)
|
|
}
|
|
|
|
log.Printf("running agent_id=%s streamd=%s data=%s workers=%d poll=%s",
|
|
cfg.AgentID, cfg.StreamdURL, cfg.DataDir, cfg.Workers, cfg.PollInterval())
|
|
log.Printf("ws heartbeat enabled — claims auto-expire if this agent dies")
|
|
log.Printf("waiting for down / missing-credential streams… Ctrl+C to stop")
|
|
|
|
ch := make(chan os.Signal, 1)
|
|
signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
|
|
<-ch
|
|
log.Printf("shutting down…")
|
|
ws.Stop()
|
|
close(stop)
|
|
}
|
|
|
|
func statusCmd(sub string, args []string) {
|
|
cfg, _ := loadCfg(sub, args)
|
|
q := openQueue(cfg)
|
|
defer q.Close()
|
|
jobs, err := q.ListRecent(40)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
tw := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0)
|
|
fmt.Fprintln(tw, "ID\tSTREAM\tAPP\tCHANNEL\tSTATUS\tATTEMPTS\tDEVICE\tREASON\tERROR")
|
|
for _, j := range jobs {
|
|
fmt.Fprintf(tw, "%d\t%s\t%s\t%s\t%s\t%d/%d\t%s\t%s\t%s\n",
|
|
j.ID, j.StreamName, j.App, j.Channel, j.Status, j.Attempts, j.MaxAttempts,
|
|
j.DeviceSerial, j.Reason, trunc(j.LastError, 60))
|
|
}
|
|
_ = tw.Flush()
|
|
}
|
|
|
|
func devicesCmd(sub string, args []string) {
|
|
cfg, _ := loadCfg(sub, args)
|
|
pool := device.New("", cfg.Devices)
|
|
list, err := pool.List()
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
if len(list) == 0 {
|
|
fmt.Println("no adb devices")
|
|
return
|
|
}
|
|
tw := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0)
|
|
fmt.Fprintln(tw, "SERIAL\tSTATE\tMODEL\tALLOWED\tBUSY\tBUSY_FOR")
|
|
for _, d := range list {
|
|
fmt.Fprintf(tw, "%s\t%s\t%s\t%v\t%v\t%s\n",
|
|
d.Serial, d.State, d.Model, d.Allowed, d.Busy, d.BusyFor)
|
|
}
|
|
_ = tw.Flush()
|
|
}
|
|
|
|
func enqueueCmd(args []string) {
|
|
fs := flag.NewFlagSet("enqueue", flag.ExitOnError)
|
|
cfgPath := fs.String("config", "", "path to apps/agent/agent.yaml")
|
|
name := fs.String("stream", "", "streamd stream name")
|
|
reason := fs.String("reason", "manual", "job reason")
|
|
_ = fs.Parse(args)
|
|
if *name == "" {
|
|
log.Fatal("--stream required")
|
|
}
|
|
cfg, err := config.Load(*cfgPath)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
api := client.New(cfg.StreamdURL, cfg.Token)
|
|
streams, err := api.ListStreams()
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
var st *client.Stream
|
|
for i := range streams {
|
|
if streams[i].Name == *name {
|
|
st = &streams[i]
|
|
break
|
|
}
|
|
}
|
|
if st == nil {
|
|
log.Fatalf("stream %q not found on streamd", *name)
|
|
}
|
|
q := openQueue(cfg)
|
|
defer q.Close()
|
|
j, added, err := q.EnqueueIfIdle(st.ID, st.Name, st.App, st.Channel, *reason, cfg.MaxAttempts)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
if !added {
|
|
fmt.Printf("already has queued/running job for %s\n", st.Name)
|
|
return
|
|
}
|
|
fmt.Printf("enqueued job=%d stream=%s app=%s channel=%s\n", j.ID, j.StreamName, j.App, j.Channel)
|
|
}
|
|
|
|
func cancelCmd(args []string) {
|
|
fs := flag.NewFlagSet("cancel", flag.ExitOnError)
|
|
cfgPath := fs.String("config", "", "path to apps/agent/agent.yaml")
|
|
jobID := fs.String("job", "", "job id")
|
|
_ = fs.Parse(args)
|
|
if *jobID == "" {
|
|
log.Fatal("--job required")
|
|
}
|
|
id, err := strconv.ParseInt(*jobID, 10, 64)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
cfg, err := config.Load(*cfgPath)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
q := openQueue(cfg)
|
|
defer q.Close()
|
|
if err := q.Cancel(id); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
fmt.Println("cancelled", id)
|
|
}
|
|
|
|
func trunc(s string, n int) string {
|
|
if len(s) <= n {
|
|
return s
|
|
}
|
|
return s[:n] + "…"
|
|
}
|