Compare commits

...

6 commits
v0.2.0 ... main

Author SHA1 Message Date
世界
93769bd46a
Update dependencies 2025-02-06 09:07:03 +08:00
世界
54badfa885
Update network handler usages 2024-11-09 12:30:16 +08:00
世界
5c2a0c4fa6
Fix accept stream 2024-11-09 12:24:55 +08:00
世界
3e7b3cfc51
Update dependencies 2024-11-09 12:23:03 +08:00
世界
6d70ef996f
Update workflows 2024-10-20 14:46:59 +08:00
世界
9bfb33698b
Fix h2mux open 2024-01-24 11:43:17 +08:00
16 changed files with 242 additions and 120 deletions

19
.github/renovate.json vendored Normal file
View file

@ -0,0 +1,19 @@
{
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
"commitMessagePrefix": "[dependencies]",
"extends": [
"config:base",
":disableRateLimiting"
],
"golang": {
"enabled": false
},
"packageRules": [
{
"matchManagers": [
"github-actions"
],
"groupName": "github-actions"
}
]
}

View file

@ -1,43 +0,0 @@
name: Debug build
on:
push:
branches:
- main
paths-ignore:
- '**.md'
- '.github/**'
- '!.github/workflows/debug.yml'
pull_request:
branches:
- main
jobs:
build:
name: Debug build
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v3
with:
fetch-depth: 0
- name: Get latest go version
id: version
run: |
echo ::set-output name=go_version::$(curl -s https://raw.githubusercontent.com/actions/go-versions/main/versions-manifest.json | grep -oE '"version": "[0-9]{1}.[0-9]{1,}(.[0-9]{1,})?"' | head -1 | cut -d':' -f2 | sed 's/ //g; s/"//g')
- name: Setup Go
uses: actions/setup-go@v4
with:
go-version: ${{ steps.version.outputs.go_version }}
- name: Add cache to Go proxy
run: |
version=`git rev-parse HEAD`
mkdir build
pushd build
go mod init build
go get -v github.com/sagernet/sing-mux@$version
popd
continue-on-error: true
- name: Build
run: |
make test

View file

@ -1,8 +1,9 @@
name: Lint name: lint
on: on:
push: push:
branches: branches:
- main
- dev - dev
paths-ignore: paths-ignore:
- '**.md' - '**.md'
@ -10,6 +11,7 @@ on:
- '!.github/workflows/lint.yml' - '!.github/workflows/lint.yml'
pull_request: pull_request:
branches: branches:
- main
- dev - dev
jobs: jobs:
@ -18,24 +20,21 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: Checkout - name: Checkout
uses: actions/checkout@v3 uses: actions/checkout@v4
with: with:
fetch-depth: 0 fetch-depth: 0
- name: Get latest go version
id: version
run: |
echo ::set-output name=go_version::$(curl -s https://raw.githubusercontent.com/actions/go-versions/main/versions-manifest.json | grep -oE '"version": "[0-9]{1}.[0-9]{1,}(.[0-9]{1,})?"' | head -1 | cut -d':' -f2 | sed 's/ //g; s/"//g')
- name: Setup Go - name: Setup Go
uses: actions/setup-go@v4 uses: actions/setup-go@v5
with: with:
go-version: ${{ steps.version.outputs.go_version }} go-version: ^1.23
- name: Cache go module - name: Cache go module
uses: actions/cache@v3 uses: actions/cache@v4
with: with:
path: | path: |
~/go/pkg/mod ~/go/pkg/mod
key: go-${{ hashFiles('**/go.sum') }} key: go-${{ hashFiles('**/go.sum') }}
- name: golangci-lint - name: golangci-lint
uses: golangci/golangci-lint-action@v3 uses: golangci/golangci-lint-action@v6
with: with:
version: latest version: latest
args: .

112
.github/workflows/test.yml vendored Normal file
View file

@ -0,0 +1,112 @@
name: test
on:
push:
branches:
- main
- dev
paths-ignore:
- '**.md'
- '.github/**'
- '!.github/workflows/debug.yml'
pull_request:
branches:
- main
- dev
jobs:
build:
name: Linux
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ^1.23
- name: Build
run: |
make test
build_go120:
name: Linux (Go 1.20)
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ~1.20
continue-on-error: true
- name: Build
run: |
make test
build_go121:
name: Linux (Go 1.21)
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ~1.21
continue-on-error: true
- name: Build
run: |
make test
build_go122:
name: Linux (Go 1.22)
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ~1.22
continue-on-error: true
- name: Build
run: |
make test
build_windows:
name: Windows
runs-on: windows-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ^1.23
continue-on-error: true
- name: Build
run: |
make test
build_darwin:
name: macOS
runs-on: macos-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ^1.23
continue-on-error: true
- name: Build
run: |
make test

1
.gitignore vendored
View file

@ -1,2 +1,3 @@
/.idea/ /.idea/
/vendor/ /vendor/
.DS_Store

View file

@ -5,6 +5,8 @@ linters:
- govet - govet
- gci - gci
- staticcheck - staticcheck
- paralleltest
- ineffassign
linters-settings: linters-settings:
gci: gci:
@ -13,5 +15,6 @@ linters-settings:
- standard - standard
- prefix(github.com/sagernet/) - prefix(github.com/sagernet/)
- default - default
staticcheck:
go: '1.20' run:
go: "1.23"

View file

@ -8,14 +8,14 @@ fmt_install:
go install -v github.com/daixiang0/gci@latest go install -v github.com/daixiang0/gci@latest
lint: lint:
GOOS=linux golangci-lint run ./... GOOS=linux golangci-lint run
GOOS=android golangci-lint run ./... GOOS=android golangci-lint run
GOOS=windows golangci-lint run ./... GOOS=windows golangci-lint run
GOOS=darwin golangci-lint run ./... GOOS=darwin golangci-lint run
GOOS=freebsd golangci-lint run ./... GOOS=freebsd golangci-lint run
lint_install: lint_install:
go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest
test: test:
go test -v ./... go test ./...

View file

@ -7,7 +7,7 @@ import (
"github.com/sagernet/sing/common" "github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/buf" "github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions" E "github.com/sagernet/sing/common/exceptions"
"github.com/sagernet/sing/common/rw" "github.com/sagernet/sing/common/varbin"
) )
const ( const (
@ -32,7 +32,7 @@ func WriteBrutalResponse(writer io.Writer, receiveBPS uint64, ok bool, message s
if ok { if ok {
common.Must(binary.Write(buffer, binary.BigEndian, receiveBPS)) common.Must(binary.Write(buffer, binary.BigEndian, receiveBPS))
} else { } else {
err := rw.WriteVString(buffer, message) err := varbin.Write(buffer, binary.BigEndian, message)
if err != nil { if err != nil {
return err return err
} }
@ -52,7 +52,7 @@ func ReadBrutalResponse(reader io.Reader) (uint64, error) {
return receiveBPS, err return receiveBPS, err
} else { } else {
var message string var message string
message, err = rw.ReadVString(reader) message, err = varbin.ReadValue[string](reader, binary.BigEndian)
if err != nil { if err != nil {
return 0, err return 0, err
} }

View file

@ -113,7 +113,7 @@ func (c *Client) openStream(ctx context.Context) (net.Conn, error) {
if err != nil { if err != nil {
continue continue
} }
stream, err = session.OpenContext(ctx) stream, err = session.Open()
if err != nil { if err != nil {
continue continue
} }
@ -206,7 +206,7 @@ func (c *Client) offerNew(ctx context.Context) (abstractSession, error) {
} }
func (c *Client) brutalExchange(ctx context.Context, sessionConn net.Conn, session abstractSession) error { func (c *Client) brutalExchange(ctx context.Context, sessionConn net.Conn, session abstractSession) error {
stream, err := session.OpenContext(ctx) stream, err := session.Open()
if err != nil { if err != nil {
return err return err
} }

10
go.mod
View file

@ -3,11 +3,11 @@ module github.com/sagernet/sing-mux
go 1.18 go 1.18
require ( require (
github.com/hashicorp/yamux v0.1.1 github.com/hashicorp/yamux v0.1.2
github.com/sagernet/sing v0.3.0 github.com/sagernet/sing v0.6.0
github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7 github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7
golang.org/x/net v0.19.0 golang.org/x/net v0.34.0
golang.org/x/sys v0.16.0 golang.org/x/sys v0.30.0
) )
require golang.org/x/text v0.14.0 // indirect require golang.org/x/text v0.21.0 // indirect

22
go.sum
View file

@ -1,18 +1,18 @@
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/hashicorp/yamux v0.1.1 h1:yrQxtgseBDrq9Y652vSRDvsKCJKOUD+GzTS4Y0Y8pvE= github.com/hashicorp/yamux v0.1.2 h1:XtB8kyFOyHXYVFnwT5C3+Bdo8gArse7j2AQ0DA0Uey8=
github.com/hashicorp/yamux v0.1.1/go.mod h1:CtWFDAQgb7dxtzFs4tWbplKIe2jSi3+5vKbgIO0SLnQ= github.com/hashicorp/yamux v0.1.2/go.mod h1:C+zze2n6e/7wshOZep2A70/aQU6QBRWJO/G6FT1wIns=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/sagernet/sing v0.2.18/go.mod h1:OL6k2F0vHmEzXz2KW19qQzu172FDgSbUSODylighuVo= github.com/sagernet/sing v0.2.18/go.mod h1:OL6k2F0vHmEzXz2KW19qQzu172FDgSbUSODylighuVo=
github.com/sagernet/sing v0.3.0 h1:PIDVFZHnQAAYRL1UYqNM+0k5s8f/tb1lUW6UDcQiOc8= github.com/sagernet/sing v0.6.0 h1:jT55zAXrG7H3x+s/FlrC15xQy3LcmuZ2GGA9+8IJdt0=
github.com/sagernet/sing v0.3.0/go.mod h1:9pfuAH6mZfgnz/YjP6xu5sxx882rfyjpcrTdUpd6w3g= github.com/sagernet/sing v0.6.0/go.mod h1:ARkL0gM13/Iv5VCZmci/NuoOlePoIsW0m7BWfln/Hak=
github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7 h1:DImB4lELfQhplLTxeq2z31Fpv8CQqqrUwTbrIRumZqQ= github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7 h1:DImB4lELfQhplLTxeq2z31Fpv8CQqqrUwTbrIRumZqQ=
github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7/go.mod h1:FP9X2xjT/Az1EsG/orYYoC+5MojWnuI7hrffz8fGwwo= github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7/go.mod h1:FP9X2xjT/Az1EsG/orYYoC+5MojWnuI7hrffz8fGwwo=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
golang.org/x/net v0.19.0 h1:zTwKpTd2XuCqf8huc7Fo2iSy+4RHPd10s4KzeTnVr1c= golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
golang.org/x/net v0.19.0/go.mod h1:CfAk/cbD4CthTvqiEl8NpboMuiuOYsAr/7NOjZJtv1U= golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU= golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

View file

@ -64,7 +64,7 @@ func (s *h2MuxServerSession) ServeHTTP(writer http.ResponseWriter, request *http
} }
} }
func (s *h2MuxServerSession) OpenContext(ctx context.Context) (net.Conn, error) { func (s *h2MuxServerSession) Open() (net.Conn, error) {
return nil, os.ErrInvalid return nil, os.ErrInvalid
} }
@ -197,7 +197,7 @@ func (s *h2MuxClientSession) MarkDead(conn *http2.ClientConn) {
s.Close() s.Close()
} }
func (s *h2MuxClientSession) OpenContext(ctx context.Context) (net.Conn, error) { func (s *h2MuxClientSession) Open() (net.Conn, error) {
pipeInReader, pipeInWriter := io.Pipe() pipeInReader, pipeInWriter := io.Pipe()
request := &http.Request{ request := &http.Request{
Method: http.MethodConnect, Method: http.MethodConnect,
@ -212,7 +212,7 @@ func (s *h2MuxClientSession) OpenContext(ctx context.Context) (net.Conn, error)
select { select {
case <-requestDone: case <-requestDone:
return return
case <-ctx.Done(): case <-time.After(TCPTimeout):
cancel() cancel()
} }
}() }()

View file

@ -12,6 +12,7 @@ import (
M "github.com/sagernet/sing/common/metadata" M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network" N "github.com/sagernet/sing/common/network"
"github.com/sagernet/sing/common/rw" "github.com/sagernet/sing/common/rw"
"github.com/sagernet/sing/common/varbin"
) )
const ( const (
@ -41,14 +42,18 @@ type Request struct {
} }
func ReadRequest(reader io.Reader) (*Request, error) { func ReadRequest(reader io.Reader) (*Request, error) {
version, err := rw.ReadByte(reader) var (
version byte
protocol byte
)
err := binary.Read(reader, binary.BigEndian, &version)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if version < Version0 || version > Version1 { if version < Version0 || version > Version1 {
return nil, E.New("unsupported version: ", version) return nil, E.New("unsupported version: ", version)
} }
protocol, err := rw.ReadByte(reader) err = binary.Read(reader, binary.BigEndian, &protocol)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -166,13 +171,12 @@ type StreamResponse struct {
func ReadStreamResponse(reader io.Reader) (*StreamResponse, error) { func ReadStreamResponse(reader io.Reader) (*StreamResponse, error) {
var response StreamResponse var response StreamResponse
status, err := rw.ReadByte(reader) err := binary.Read(reader, binary.BigEndian, &response.Status)
if err != nil { if err != nil {
return nil, err return nil, err
} }
response.Status = status if response.Status == statusError {
if status == statusError { response.Message, err = varbin.ReadValue[string](reader, binary.BigEndian)
response.Message, err = rw.ReadVString(reader)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View file

@ -13,15 +13,24 @@ import (
"github.com/sagernet/sing/common/task" "github.com/sagernet/sing/common/task"
) )
// Deprecated: Use ServiceHandlerEx instead.
//
//nolint:staticcheck
type ServiceHandler interface { type ServiceHandler interface {
N.TCPConnectionHandler N.TCPConnectionHandler
N.UDPConnectionHandler N.UDPConnectionHandler
} }
type ServiceHandlerEx interface {
N.TCPConnectionHandlerEx
N.UDPConnectionHandlerEx
}
type Service struct { type Service struct {
newStreamContext func(context.Context, net.Conn) context.Context newStreamContext func(context.Context, net.Conn) context.Context
logger logger.ContextLogger logger logger.ContextLogger
handler ServiceHandler handler ServiceHandler
handlerEx ServiceHandlerEx
padding bool padding bool
brutal BrutalOptions brutal BrutalOptions
} }
@ -30,6 +39,7 @@ type ServiceOptions struct {
NewStreamContext func(context.Context, net.Conn) context.Context NewStreamContext func(context.Context, net.Conn) context.Context
Logger logger.ContextLogger Logger logger.ContextLogger
Handler ServiceHandler Handler ServiceHandler
HandlerEx ServiceHandlerEx
Padding bool Padding bool
Brutal BrutalOptions Brutal BrutalOptions
} }
@ -42,12 +52,26 @@ func NewService(options ServiceOptions) (*Service, error) {
newStreamContext: options.NewStreamContext, newStreamContext: options.NewStreamContext,
logger: options.Logger, logger: options.Logger,
handler: options.Handler, handler: options.Handler,
handlerEx: options.HandlerEx,
padding: options.Padding, padding: options.Padding,
brutal: options.Brutal, brutal: options.Brutal,
}, nil }, nil
} }
// Deprecated: Use NewConnectionEx instead.
func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error { func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
return s.newConnection(ctx, conn, metadata.Source)
}
func (s *Service) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
err := s.newConnection(ctx, conn, source)
N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil {
s.logger.ErrorContext(ctx, E.Cause(err, "process multiplex connection from ", source))
}
}
func (s *Service) newConnection(ctx context.Context, conn net.Conn, source M.Socksaddr) error {
request, err := ReadRequest(conn) request, err := ReadRequest(conn)
if err != nil { if err != nil {
return err return err
@ -63,17 +87,17 @@ func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.M
} }
var group task.Group var group task.Group
group.Append0(func(_ context.Context) error { group.Append0(func(_ context.Context) error {
var stream net.Conn
for { for {
stream, err = session.Accept() stream, aErr := session.Accept()
if err != nil { if aErr != nil {
return err return aErr
} }
streamCtx := s.newStreamContext(ctx, stream) streamCtx := s.newStreamContext(ctx, stream)
go func() { go func() {
hErr := s.newConnection(streamCtx, conn, stream, metadata) hErr := s.newSession(streamCtx, conn, stream, source)
if hErr != nil { if hErr != nil {
s.logger.ErrorContext(streamCtx, E.Cause(hErr, "handle connection")) stream.Close()
s.logger.ErrorContext(streamCtx, E.Cause(hErr, "process multiplex stream"))
} }
}() }()
} }
@ -84,13 +108,13 @@ func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.M
return group.Run(ctx) return group.Run(ctx)
} }
func (s *Service) newConnection(ctx context.Context, sessionConn net.Conn, stream net.Conn, metadata M.Metadata) error { func (s *Service) newSession(ctx context.Context, sessionConn net.Conn, stream net.Conn, source M.Socksaddr) error {
stream = &wrapStream{stream} stream = &wrapStream{stream}
request, err := ReadStreamRequest(stream) request, err := ReadStreamRequest(stream)
if err != nil { if err != nil {
return E.Cause(err, "read multiplex stream request") return E.Cause(err, "read multiplex stream request")
} }
metadata.Destination = request.Destination destination := request.Destination
if request.Network == N.NetworkTCP { if request.Network == N.NetworkTCP {
conn := &serverConn{ExtendedConn: bufio.NewExtendedConn(stream)} conn := &serverConn{ExtendedConn: bufio.NewExtendedConn(stream)}
if request.Destination.Fqdn == BrutalExchangeDomain { if request.Destination.Fqdn == BrutalExchangeDomain {
@ -128,20 +152,28 @@ func (s *Service) newConnection(ctx context.Context, sessionConn net.Conn, strea
} }
return nil return nil
} }
s.logger.InfoContext(ctx, "inbound multiplex connection to ", metadata.Destination) s.logger.InfoContext(ctx, "inbound multiplex connection to ", destination)
s.handler.NewConnection(ctx, conn, metadata) if s.handler != nil {
stream.Close() //nolint:staticcheck
s.handler.NewConnection(ctx, conn, M.Metadata{Source: source, Destination: destination})
} else {
s.handlerEx.NewConnectionEx(ctx, conn, source, destination, nil)
}
} else { } else {
var packetConn N.PacketConn var packetConn N.PacketConn
if !request.PacketAddr { if !request.PacketAddr {
s.logger.InfoContext(ctx, "inbound multiplex packet connection to ", metadata.Destination) s.logger.InfoContext(ctx, "inbound multiplex packet connection to ", destination)
packetConn = &serverPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: request.Destination} packetConn = &serverPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: request.Destination}
} else { } else {
s.logger.InfoContext(ctx, "inbound multiplex packet connection") s.logger.InfoContext(ctx, "inbound multiplex packet connection")
packetConn = &serverPacketAddrConn{ExtendedConn: bufio.NewExtendedConn(stream)} packetConn = &serverPacketAddrConn{ExtendedConn: bufio.NewExtendedConn(stream)}
} }
s.handler.NewPacketConnection(ctx, packetConn, metadata) if s.handler != nil {
stream.Close() //nolint:staticcheck
s.handler.NewPacketConnection(ctx, packetConn, M.Metadata{Source: source, Destination: destination})
} else {
s.handlerEx.NewPacketConnectionEx(ctx, packetConn, source, destination, nil)
}
} }
return nil return nil
} }

View file

@ -10,7 +10,7 @@ import (
"github.com/sagernet/sing/common/buf" "github.com/sagernet/sing/common/buf"
M "github.com/sagernet/sing/common/metadata" M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network" N "github.com/sagernet/sing/common/network"
"github.com/sagernet/sing/common/rw" "github.com/sagernet/sing/common/varbin"
) )
type serverConn struct { type serverConn struct {
@ -24,11 +24,11 @@ func (c *serverConn) NeedHandshake() bool {
func (c *serverConn) HandshakeFailure(err error) error { func (c *serverConn) HandshakeFailure(err error) error {
errMessage := err.Error() errMessage := err.Error()
buffer := buf.NewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage)) buffer := buf.NewSize(1 + varbin.UvarintLen(uint64(len(errMessage))) + len(errMessage))
defer buffer.Release() defer buffer.Release()
common.Must( common.Must(
buffer.WriteByte(statusError), buffer.WriteByte(statusError),
rw.WriteVString(buffer, errMessage), varbin.Write(buffer, binary.BigEndian, errMessage),
) )
return common.Error(c.ExtendedConn.Write(buffer.Bytes())) return common.Error(c.ExtendedConn.Write(buffer.Bytes()))
} }
@ -88,11 +88,11 @@ func (c *serverPacketConn) NeedHandshake() bool {
func (c *serverPacketConn) HandshakeFailure(err error) error { func (c *serverPacketConn) HandshakeFailure(err error) error {
errMessage := err.Error() errMessage := err.Error()
buffer := buf.NewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage)) buffer := buf.NewSize(1 + varbin.UvarintLen(uint64(len(errMessage))) + len(errMessage))
defer buffer.Release() defer buffer.Release()
common.Must( common.Must(
buffer.WriteByte(statusError), buffer.WriteByte(statusError),
rw.WriteVString(buffer, errMessage), varbin.Write(buffer, binary.BigEndian, errMessage),
) )
return common.Error(c.ExtendedConn.Write(buffer.Bytes())) return common.Error(c.ExtendedConn.Write(buffer.Bytes()))
} }
@ -188,11 +188,11 @@ func (c *serverPacketAddrConn) NeedHandshake() bool {
func (c *serverPacketAddrConn) HandshakeFailure(err error) error { func (c *serverPacketAddrConn) HandshakeFailure(err error) error {
errMessage := err.Error() errMessage := err.Error()
buffer := buf.NewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage)) buffer := buf.NewSize(1 + varbin.UvarintLen(uint64(len(errMessage))) + len(errMessage))
defer buffer.Release() defer buffer.Release()
common.Must( common.Must(
buffer.WriteByte(statusError), buffer.WriteByte(statusError),
rw.WriteVString(buffer, errMessage), varbin.Write(buffer, binary.BigEndian, errMessage),
) )
return common.Error(c.ExtendedConn.Write(buffer.Bytes())) return common.Error(c.ExtendedConn.Write(buffer.Bytes()))
} }

View file

@ -1,7 +1,6 @@
package mux package mux
import ( import (
"context"
"io" "io"
"net" "net"
"reflect" "reflect"
@ -13,7 +12,7 @@ import (
) )
type abstractSession interface { type abstractSession interface {
OpenContext(ctx context.Context) (net.Conn, error) Open() (net.Conn, error)
Accept() (net.Conn, error) Accept() (net.Conn, error)
NumStreams() int NumStreams() int
Close() error Close() error
@ -81,7 +80,7 @@ type smuxSession struct {
*smux.Session *smux.Session
} }
func (s *smuxSession) OpenContext(context.Context) (net.Conn, error) { func (s *smuxSession) Open() (net.Conn, error) {
return s.OpenStream() return s.OpenStream()
} }
@ -97,10 +96,6 @@ type yamuxSession struct {
*yamux.Session *yamux.Session
} }
func (y *yamuxSession) OpenContext(context.Context) (net.Conn, error) {
return y.OpenStream()
}
func (y *yamuxSession) CanTakeNewRequest() bool { func (y *yamuxSession) CanTakeNewRequest() bool {
return true return true
} }