|
| 1 | +package dsi |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "io" |
| 6 | + "net" |
| 7 | + "sync" |
| 8 | + |
| 9 | + "github.com/ObsoleteMadness/ClassicStack/core/component" |
| 10 | + dsiproto "github.com/ObsoleteMadness/ClassicStack/core/protocol/dsi" |
| 11 | + "github.com/ObsoleteMadness/ClassicStack/core/service/afp" |
| 12 | + |
| 13 | + "github.com/ObsoleteMadness/ClassicStack/core/log" |
| 14 | +) |
| 15 | + |
| 16 | +// Name is the component name for the AFP-over-TCP (DSI) transport. It is its own |
| 17 | +// supervised component (a listener with a lifecycle), distinct from the AFP command |
| 18 | +// service. |
| 19 | +const Name = "DSI" |
| 20 | + |
| 21 | +// maxMessage caps a single DSI data block at 16 MiB — well above any real AFP |
| 22 | +// command/write payload — so a malformed DataLen header cannot drive an unbounded |
| 23 | +// allocation. |
| 24 | +const maxMessage = 16 << 20 |
| 25 | + |
| 26 | +// Transport is a TCP listener that drives the AFP command-core seam over DSI framing. |
| 27 | +// One accept loop spawns a goroutine per connection; each connection opens one AFP |
| 28 | +// circuit (on OpenSession) and serves DSI requests until the peer closes or sends |
| 29 | +// CloseSession. |
| 30 | +type Transport struct { |
| 31 | + addr string |
| 32 | + handler afp.CommandHandler |
| 33 | + logger log.Logger |
| 34 | + |
| 35 | + mu sync.Mutex |
| 36 | + listener net.Listener |
| 37 | + conns map[net.Conn]struct{} |
| 38 | + running bool |
| 39 | +} |
| 40 | + |
| 41 | +// New builds a DSI transport. addr/handler may be empty/nil at construction (the |
| 42 | +// registry builds it inert); the compose transport cross-wire installs the AFP |
| 43 | +// command handler and the listen address once the AFP service and its tcp_addr are |
| 44 | +// resolved (mirrors adapter/smbtcp.New). |
| 45 | +func New(addr string, handler afp.CommandHandler, logger log.Logger) *Transport { |
| 46 | + return &Transport{addr: addr, handler: handler, logger: logger, conns: make(map[net.Conn]struct{})} |
| 47 | +} |
| 48 | + |
| 49 | +// SetHandler installs the AFP command handler after construction. Must be called |
| 50 | +// before Start; a nil handler leaves Start a no-op. |
| 51 | +func (t *Transport) SetHandler(h afp.CommandHandler) { |
| 52 | + t.mu.Lock() |
| 53 | + t.handler = h |
| 54 | + t.mu.Unlock() |
| 55 | +} |
| 56 | + |
| 57 | +// SetAddr sets/overrides the listen address before Start (compose supplies it from |
| 58 | +// the AFP server section's tcp_addr). An empty address keeps Start a no-op. |
| 59 | +func (t *Transport) SetAddr(addr string) { |
| 60 | + t.mu.Lock() |
| 61 | + t.addr = addr |
| 62 | + t.mu.Unlock() |
| 63 | +} |
| 64 | + |
| 65 | +// Name returns the component name. |
| 66 | +func (t *Transport) Name() string { return Name } |
| 67 | + |
| 68 | +// Binding reports the listen address (component.Bindable), so the dashboard shows it. |
| 69 | +func (t *Transport) Binding() string { return t.addr } |
| 70 | + |
| 71 | +// Dependencies declares the DSI listener's start-order edge: the AFP service must be |
| 72 | +// running first, since the listener drives its command-core seam (and must stop |
| 73 | +// before it). Drops in a build without the AFP service. |
| 74 | +func (t *Transport) Dependencies() []string { return []string{afp.Name} } |
| 75 | + |
| 76 | +// Start opens the listener and begins accepting. Idempotent (§3). A nil handler or an |
| 77 | +// empty address makes Start a no-op so a build that wires the transport but does not |
| 78 | +// configure tcp_addr stays inert rather than erroring. |
| 79 | +// |
| 80 | +// A bind failure is NON-FATAL, matching the other transports' graceful-degradation |
| 81 | +// posture: Start logs a warning and returns nil rather than aborting the whole |
| 82 | +// stack's bring-up. |
| 83 | +func (t *Transport) Start(_ context.Context) error { |
| 84 | + t.mu.Lock() |
| 85 | + defer t.mu.Unlock() |
| 86 | + if t.running || t.handler == nil || t.addr == "" { |
| 87 | + return nil |
| 88 | + } |
| 89 | + l, err := net.Listen("tcp", t.addr) |
| 90 | + if err != nil { |
| 91 | + if t.logger != nil { |
| 92 | + t.logger.Log(log.Warn, "AFP-over-TCP (DSI) bind failed; transport inert", |
| 93 | + log.Str("addr", t.addr), log.Str("error", err.Error())) |
| 94 | + } |
| 95 | + t.running = true // lifecycle-consistent: "running" but unbound |
| 96 | + return nil |
| 97 | + } |
| 98 | + t.listener = l |
| 99 | + t.running = true |
| 100 | + go t.acceptLoop(l) |
| 101 | + if t.logger != nil { |
| 102 | + t.logger.Log(log.Info, "AFP-over-TCP (DSI) listening", log.Str("addr", l.Addr().String())) |
| 103 | + } |
| 104 | + return nil |
| 105 | +} |
| 106 | + |
| 107 | +// Stop closes the listener and every live connection. Safe after a partial Start (§3). |
| 108 | +func (t *Transport) Stop(_ context.Context) error { |
| 109 | + t.mu.Lock() |
| 110 | + if !t.running { |
| 111 | + t.mu.Unlock() |
| 112 | + return nil |
| 113 | + } |
| 114 | + t.running = false |
| 115 | + l := t.listener |
| 116 | + t.listener = nil |
| 117 | + conns := make([]net.Conn, 0, len(t.conns)) |
| 118 | + for c := range t.conns { |
| 119 | + conns = append(conns, c) |
| 120 | + } |
| 121 | + t.mu.Unlock() |
| 122 | + |
| 123 | + if l != nil { |
| 124 | + _ = l.Close() |
| 125 | + } |
| 126 | + for _, c := range conns { |
| 127 | + _ = c.Close() |
| 128 | + } |
| 129 | + return nil |
| 130 | +} |
| 131 | + |
| 132 | +func (t *Transport) acceptLoop(l net.Listener) { |
| 133 | + for { |
| 134 | + conn, err := l.Accept() |
| 135 | + if err != nil { |
| 136 | + return // listener closed (Stop) or a fatal accept error |
| 137 | + } |
| 138 | + t.mu.Lock() |
| 139 | + if !t.running { |
| 140 | + t.mu.Unlock() |
| 141 | + _ = conn.Close() |
| 142 | + return |
| 143 | + } |
| 144 | + t.conns[conn] = struct{}{} |
| 145 | + t.mu.Unlock() |
| 146 | + go t.serve(conn) |
| 147 | + } |
| 148 | +} |
| 149 | + |
| 150 | +// serve runs one connection: answer sessionless GetStatus directly, open an AFP |
| 151 | +// circuit on OpenSession, dispatch Command/Write through it, and close the circuit on |
| 152 | +// CloseSession or when the peer disconnects. |
| 153 | +func (t *Transport) serve(conn net.Conn) { |
| 154 | + var circuit afp.CommandCircuit |
| 155 | + defer func() { |
| 156 | + if circuit != nil { |
| 157 | + circuit.Close() |
| 158 | + } |
| 159 | + _ = conn.Close() |
| 160 | + t.mu.Lock() |
| 161 | + delete(t.conns, conn) |
| 162 | + t.mu.Unlock() |
| 163 | + }() |
| 164 | + |
| 165 | + handler := t.handlerRef() |
| 166 | + hdrBuf := make([]byte, dsiproto.HeaderSize) |
| 167 | + for { |
| 168 | + if _, err := io.ReadFull(conn, hdrBuf); err != nil { |
| 169 | + return |
| 170 | + } |
| 171 | + var h dsiproto.Header |
| 172 | + if !h.Unmarshal(hdrBuf) { |
| 173 | + return |
| 174 | + } |
| 175 | + if h.DataLen > maxMessage { |
| 176 | + return |
| 177 | + } |
| 178 | + payload := make([]byte, h.DataLen) |
| 179 | + if h.DataLen > 0 { |
| 180 | + if _, err := io.ReadFull(conn, payload); err != nil { |
| 181 | + return |
| 182 | + } |
| 183 | + } |
| 184 | + |
| 185 | + switch h.Command { |
| 186 | + case dsiproto.GetStatus: |
| 187 | + t.reply(conn, h.RequestID, dsiproto.GetStatus, 0, handler.GetServerInfo()) |
| 188 | + case dsiproto.OpenSession: |
| 189 | + if circuit != nil { |
| 190 | + circuit.Close() |
| 191 | + } |
| 192 | + circuit = handler.NewConn() |
| 193 | + t.reply(conn, h.RequestID, dsiproto.OpenSession, 0, nil) |
| 194 | + case dsiproto.Command, dsiproto.Write: |
| 195 | + if circuit == nil { |
| 196 | + // A Command/Write before OpenSession is a protocol violation; there is |
| 197 | + // no AFP result code for "no session" (that is a DSI-level concern), so |
| 198 | + // the connection is simply dropped, matching how the ATP spine answers |
| 199 | + // an unknown ASP session id with a hard error rather than serving. |
| 200 | + return |
| 201 | + } |
| 202 | + reply, result := circuit.Command(payload) |
| 203 | + t.reply(conn, h.RequestID, h.Command, uint32(result), reply) |
| 204 | + case dsiproto.Tickle: |
| 205 | + // No reply required (mirrors ASP's SPTickle) — Tickle exists only to reset |
| 206 | + // the peer's idle timer, whichever direction it travels. |
| 207 | + case dsiproto.CloseSession: |
| 208 | + if circuit != nil { |
| 209 | + circuit.Close() |
| 210 | + circuit = nil |
| 211 | + } |
| 212 | + t.reply(conn, h.RequestID, dsiproto.CloseSession, 0, nil) |
| 213 | + return |
| 214 | + default: |
| 215 | + // Unknown command: ignore and keep the connection open, matching the old |
| 216 | + // server's tolerance of unrecognised DSI commands. |
| 217 | + } |
| 218 | + } |
| 219 | +} |
| 220 | + |
| 221 | +// reply writes one DSI reply frame. The AFP/DSI result code goes in the header's |
| 222 | +// ErrorOffset field (its reply-side "ErrorCode" role) — NOT prepended to the payload — |
| 223 | +// per the DSI header contract documented in core/protocol/dsi; see spec/21-dsi.md. |
| 224 | +func (t *Transport) reply(conn net.Conn, reqID uint16, cmd uint8, errCode uint32, data []byte) { |
| 225 | + h := dsiproto.Header{ |
| 226 | + Flags: dsiproto.Reply, |
| 227 | + Command: cmd, |
| 228 | + RequestID: reqID, |
| 229 | + ErrorOffset: errCode, |
| 230 | + DataLen: uint32(len(data)), |
| 231 | + } |
| 232 | + if _, err := conn.Write(h.Marshal()); err != nil { |
| 233 | + return |
| 234 | + } |
| 235 | + if len(data) > 0 { |
| 236 | + _, _ = conn.Write(data) |
| 237 | + } |
| 238 | +} |
| 239 | + |
| 240 | +func (t *Transport) handlerRef() afp.CommandHandler { |
| 241 | + t.mu.Lock() |
| 242 | + defer t.mu.Unlock() |
| 243 | + return t.handler |
| 244 | +} |
| 245 | + |
| 246 | +var ( |
| 247 | + _ component.Component = (*Transport)(nil) |
| 248 | + _ component.Bindable = (*Transport)(nil) |
| 249 | + _ component.DependsOn = (*Transport)(nil) |
| 250 | +) |
0 commit comments