From ff85ea083805badcc9dc55c0f7fae346f52e1121 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Tue, 8 Sep 2026 20:48:28 +0200 Subject: [PATCH] [client] Enforce a file drop block on a transfer already in flight The receiving policy was read once, in submitOffer, and never again. The upload path and the reader guarding it both watched only the offer's own Decision, so blocking a sender refused its next offer while the transfer it had already started kept streaming into the spool and was delivered. SetSenderRule tried to settle the sender's offers but went through Decide, which only moves an offer out of Pending, and skipped everything else - precisely the accepted ones that were still running. It now revokes any non-terminal offer instead, and the staged copy re-reads the policy on the same cadence it already checks the decision, so a block stops the bytes. State.terminal carries the state list that Transfer.terminal duplicated. --- client/internal/filedrop/filedrop_test.go | 56 +++++++++++++++++++++++ client/internal/filedrop/history.go | 7 +-- client/internal/filedrop/manager.go | 13 ++++-- client/internal/filedrop/protocol.go | 9 ++++ client/internal/filedrop/receiver.go | 7 ++- 5 files changed, 80 insertions(+), 12 deletions(-) diff --git a/client/internal/filedrop/filedrop_test.go b/client/internal/filedrop/filedrop_test.go index c8da0d8e1..69c7aad81 100644 --- a/client/internal/filedrop/filedrop_test.go +++ b/client/internal/filedrop/filedrop_test.go @@ -399,6 +399,62 @@ func TestOfferIsScopedToItsSender(t *testing.T) { assert.ErrorIs(t, err, ErrOfferNotFound, "another peer must not poll the offer") } +func TestBlockingASenderStopsAnAcceptedUpload(t *testing.T) { + policy := NewPolicyStore(testProfile) + require.NoError(t, policy.Set(Policy{Mode: ModeAutoAccept})) + + spool, err := NewSpool(t.TempDir()) + require.NoError(t, err) + + store := NewOfferStore(time.Minute) + r := &receiver{offers: store, policy: policy, spool: spool} + + content := strings.Repeat("x", 512<<10) + offer := store.Add(testPeer, "sender", []FileMeta{{Name: "a.bin", Size: int64(len(content))}}, DecisionAccepted) + require.NoError(t, spool.Prepare(offer.ID)) + + require.NoError(t, policy.SetSenderRule(testPeer, SenderRuleBlock)) + + err = r.upload(senderIdentity{key: testPeer}, offer.ID, 0, 0, strings.NewReader(content)) + require.ErrorIs(t, err, ErrNotAccepted, "a blocked sender must not keep uploading") + + staged, serr := spool.Received(offer.ID, 0) + require.NoError(t, serr) + assert.Less(t, staged, int64(len(content)), "the copy must stop short of the announced size") +} + +func TestBlockingASenderRevokesAnAcceptedOffer(t *testing.T) { + mgr, err := NewManager(ManagerConfig{Profile: testProfile, DataDir: t.TempDir()}) + require.NoError(t, err) + require.NoError(t, mgr.Policy().Set(Policy{Mode: ModeAutoAccept})) + + srv, err := NewServer(ServerConfig{ + SpoolDir: t.TempDir(), + Policy: mgr.Policy(), + Resolver: staticResolver{key: testPeer}, + Notifier: mgr, + }) + require.NoError(t, err) + + mgr.mu.Lock() + mgr.server = srv + mgr.mu.Unlock() + + offer := srv.Offers().Add(testPeer, "sender", []FileMeta{{Name: "a.bin", Size: 10}}, DecisionAccepted) + require.NoError(t, srv.Spool().Prepare(offer.ID)) + mgr.OnOffer(offer) + + require.NoError(t, mgr.SetSenderRule(testPeer, SenderRuleBlock)) + + current, ok := srv.Offers().Get(testPeer, offer.ID) + require.True(t, ok) + assert.NotEqual(t, DecisionAccepted, current.Decision, "an accepted offer must lose its consent") + + transfer, ok := mgr.history.Get(offer.ID) + require.True(t, ok) + assert.True(t, transfer.terminal(), "the transfer must be settled, not left running") +} + func TestPolicyEvaluation(t *testing.T) { store := NewPolicyStore(testProfile) require.NoError(t, store.Set(Policy{Mode: ModeAsk})) diff --git a/client/internal/filedrop/history.go b/client/internal/filedrop/history.go index eed2f13e7..e5ee07a3e 100644 --- a/client/internal/filedrop/history.go +++ b/client/internal/filedrop/history.go @@ -87,12 +87,7 @@ func LoadHistory(store Store) *History { } func (t Transfer) terminal() bool { - switch t.State { - case StateCompleted, StateDeclined, StateExpired, StateCancelled, StateFailed: - return true - default: - return false - } + return t.State.terminal() } func (t Transfer) clone() Transfer { diff --git a/client/internal/filedrop/manager.go b/client/internal/filedrop/manager.go index 71725a3e6..6315aac96 100644 --- a/client/internal/filedrop/manager.go +++ b/client/internal/filedrop/manager.go @@ -416,14 +416,17 @@ func (m *Manager) SetSenderRule(peer PeerKey, rule SenderRule) error { return nil } + // Revoke rather than Decide: Decide only moves an offer out of Pending, so + // on its own it would leave a transfer the sender already had accepted + // streaming into the spool after the user blocked them. for _, offer := range server.Offers().List() { - if offer.Sender != peer || offer.Decision != DecisionPending { + if offer.Sender != peer || offer.State.terminal() { continue } - if declined, ok := server.Offers().Decide(offer.ID, DecisionDeclined); ok { - server.Spool().Remove(declined.ID) - m.finishTransfer(declined.ID, StateDeclined, "") - m.emit(EventWithdrawn, m.transferOf(declined.ID)) + if revoked, ok := server.Offers().Revoke(offer.ID); ok { + server.Spool().Remove(revoked.ID) + m.finishTransfer(revoked.ID, StateDeclined, "") + m.emit(EventWithdrawn, m.transferOf(revoked.ID)) } } return nil diff --git a/client/internal/filedrop/protocol.go b/client/internal/filedrop/protocol.go index d0a65ff6e..c2eb3c045 100644 --- a/client/internal/filedrop/protocol.go +++ b/client/internal/filedrop/protocol.go @@ -201,3 +201,12 @@ func (s State) String() string { return fmt.Sprintf("unknown(%d)", uint8(s)) } } + +func (s State) terminal() bool { + switch s { + case StateCompleted, StateDeclined, StateExpired, StateCancelled, StateFailed: + return true + default: + return false + } +} diff --git a/client/internal/filedrop/receiver.go b/client/internal/filedrop/receiver.go index 926a65752..a4cea0335 100644 --- a/client/internal/filedrop/receiver.go +++ b/client/internal/filedrop/receiver.go @@ -140,10 +140,15 @@ func (r *receiver) upload(sender senderIdentity, id OfferID, index int, offset i // The decision is read once, at the top, but a whole file goes into this // one request: a receiver that declines halfway through would otherwise be - // streamed the rest of it, into a spool it has already thrown away. + // streamed the rest of it, into a spool it has already thrown away. The + // policy is re-read alongside it, so blocking the sender mid-transfer stops + // the bytes too rather than only refusing its next offer. watched := &acceptedReader{ r: body, accepted: func() bool { + if r.policy.Evaluate(sender.key) == ModeOff { + return false + } current, ok := r.offers.Get(sender.key, id) return ok && current.Decision == DecisionAccepted },