Merge branch 'master' of https://github.com/seaweedfs/seaweedfs
This commit is contained in:
Chris Lu
commit
a0be40e070
100
.github/workflows/helm_ci.yml
vendored
100
.github/workflows/helm_ci.yml
vendored
@ -217,6 +217,106 @@ jobs:
|
||||
print("✓ No blank lines in security+S3 command blocks")
|
||||
PYEOF
|
||||
|
||||
echo ""
|
||||
echo "=== Testing security+S3: -cert.file/-key.file gated on httpsPort (issue #9202) ==="
|
||||
# Regression test: when enableSecurity=true but *.httpsPort is 0 (the default),
|
||||
# the chart must NOT emit -cert.file / -key.file to the S3 frontend. Passing
|
||||
# them promotes weed s3's main -port to HTTPS (see weed/command/s3.go), which
|
||||
# makes the HTTP readinessProbe spam "TLS handshake error ... client sent an
|
||||
# HTTP request to an HTTPS server" into the pod log.
|
||||
#
|
||||
# When *.httpsPort > 0, both -port.https and cert/key args MUST be emitted
|
||||
# together so the opt-in HTTPS listener actually has credentials.
|
||||
python3 - "$CHART_DIR" <<'PYEOF'
|
||||
import subprocess, sys, yaml
|
||||
chart = sys.argv[1]
|
||||
|
||||
def render(values):
|
||||
args = ["helm", "template", "test", chart]
|
||||
for k, v in values.items():
|
||||
args += ["--set", f"{k}={v}"]
|
||||
return subprocess.check_output(args, text=True)
|
||||
|
||||
def script_of(manifest, kind_name):
|
||||
for doc in yaml.safe_load_all(manifest):
|
||||
if not doc or doc.get("kind") not in ("Deployment", "StatefulSet"):
|
||||
continue
|
||||
if doc["metadata"]["name"] != kind_name:
|
||||
continue
|
||||
for c in doc["spec"]["template"]["spec"]["containers"]:
|
||||
cmd = c.get("command", [])
|
||||
if len(cmd) >= 3 and cmd[0] == "/bin/sh" and cmd[1] == "-ec":
|
||||
return cmd[2]
|
||||
raise AssertionError(f"no container script for {kind_name}")
|
||||
|
||||
cases = [
|
||||
# (values, workload-name, httpsPort-set?, arg-prefix)
|
||||
({"global.seaweedfs.enableSecurity": "true",
|
||||
"s3.enabled": "true"},
|
||||
"test-seaweedfs-s3", False, ""),
|
||||
({"global.seaweedfs.enableSecurity": "true",
|
||||
"s3.enabled": "true",
|
||||
"s3.httpsPort": "8443"},
|
||||
"test-seaweedfs-s3", True, ""),
|
||||
({"global.seaweedfs.enableSecurity": "true",
|
||||
"filer.s3.enabled": "true"},
|
||||
"test-seaweedfs-filer", False, "s3."),
|
||||
({"global.seaweedfs.enableSecurity": "true",
|
||||
"filer.s3.enabled": "true",
|
||||
"filer.s3.httpsPort": "8444"},
|
||||
"test-seaweedfs-filer", True, "s3."),
|
||||
({"global.seaweedfs.enableSecurity": "true",
|
||||
"allInOne.enabled": "true",
|
||||
"allInOne.s3.enabled": "true"},
|
||||
"test-seaweedfs-all-in-one", False, "s3."),
|
||||
({"global.seaweedfs.enableSecurity": "true",
|
||||
"allInOne.enabled": "true",
|
||||
"allInOne.s3.enabled": "true",
|
||||
"allInOne.s3.httpsPort": "8445"},
|
||||
"test-seaweedfs-all-in-one", True, "s3."),
|
||||
]
|
||||
|
||||
failed = False
|
||||
for values, name, https_on, prefix in cases:
|
||||
script = script_of(render(values), name)
|
||||
cert_flag = f"-{prefix}cert.file="
|
||||
key_flag = f"-{prefix}key.file="
|
||||
https_flag = f"-{prefix}port.https="
|
||||
has_cert = cert_flag in script
|
||||
has_key = key_flag in script
|
||||
has_https = https_flag in script
|
||||
label = f"{name} (httpsPort {'set' if https_on else 'unset'})"
|
||||
if https_on:
|
||||
if not (has_cert and has_key and has_https):
|
||||
print(f"FAIL: {label}: expected {cert_flag}, {key_flag}, {https_flag} all present "
|
||||
f"(got cert={has_cert} key={has_key} https={has_https})", file=sys.stderr)
|
||||
failed = True
|
||||
else:
|
||||
print(f"✓ {label}: cert/key/https args emitted together")
|
||||
else:
|
||||
if has_cert or has_key or has_https:
|
||||
print(f"FAIL: {label}: expected none of {cert_flag}/{key_flag}/{https_flag}; "
|
||||
f"main S3 -port would silently become HTTPS and break HTTP probes "
|
||||
f"(got cert={has_cert} key={has_key} https={has_https})", file=sys.stderr)
|
||||
failed = True
|
||||
else:
|
||||
print(f"✓ {label}: no TLS args emitted, main -port stays HTTP")
|
||||
|
||||
# bash -n: pin down that the rendered script parses. Guards against
|
||||
# a future helper change that leaves a dangling `\` with nothing
|
||||
# after it (every current caller already exits cleanly because
|
||||
# bash treats trailing `\<newline><EOF>` as line-continuation to
|
||||
# an empty line — but keep the contract explicit).
|
||||
parse = subprocess.run(["bash", "-n"], input=script, text=True,
|
||||
capture_output=True)
|
||||
if parse.returncode != 0:
|
||||
print(f"FAIL: {label}: bash -n rejected rendered script: {parse.stderr.strip()}",
|
||||
file=sys.stderr)
|
||||
failed = True
|
||||
|
||||
sys.exit(1 if failed else 0)
|
||||
PYEOF
|
||||
|
||||
echo "✅ All template rendering tests passed!"
|
||||
|
||||
- name: Create kind cluster
|
||||
|
||||
2
go.mod
2
go.mod
@ -303,7 +303,7 @@ require (
|
||||
github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3 // indirect
|
||||
github.com/Azure/go-ntlmssp v0.1.0 // indirect
|
||||
github.com/Azure/go-ntlmssp v0.1.1 // indirect
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 // indirect
|
||||
github.com/Files-com/files-sdk-go/v3 v3.2.264 // indirect
|
||||
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 // indirect
|
||||
|
||||
4
go.sum
4
go.sum
@ -577,8 +577,8 @@ github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK
|
||||
github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
|
||||
github.com/Azure/go-autorest/autorest/to v0.4.1 h1:CxNHBqdzTr7rLtdrtb5CMjJcDut+WNGCVv7OmS5+lTc=
|
||||
github.com/Azure/go-autorest/autorest/to v0.4.1/go.mod h1:EtaofgU4zmtvn1zT2ARsjRFdq9vXx0YWtmElwL+GZ9M=
|
||||
github.com/Azure/go-ntlmssp v0.1.0 h1:DjFo6YtWzNqNvQdrwEyr/e4nhU3vRiwenz5QX7sFz+A=
|
||||
github.com/Azure/go-ntlmssp v0.1.0/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk=
|
||||
github.com/Azure/go-ntlmssp v0.1.1 h1:l+FM/EEMb0U9QZE7mKNEDw5Mu3mFiaa2GKOoTSsNDPw=
|
||||
github.com/Azure/go-ntlmssp v0.1.1/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk=
|
||||
github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM=
|
||||
github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE=
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgvJqCH0sFfrBUTnUJSBrBf7++ypk+twtRs=
|
||||
|
||||
@ -247,9 +247,9 @@ spec:
|
||||
{{- $httpsPort := .Values.allInOne.s3.httpsPort | default .Values.s3.httpsPort }}
|
||||
{{- if $httpsPort }}
|
||||
-s3.port.https={{ $httpsPort }} \
|
||||
{{- end }}
|
||||
{{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }}
|
||||
{{- end }}
|
||||
{{- end }}
|
||||
{{- if or .Values.allInOne.s3.enableAuth .Values.s3.enableAuth .Values.filer.s3.enableAuth }}
|
||||
-s3.config=/etc/sw/s3/seaweedfs_s3_config \
|
||||
{{- end }}
|
||||
|
||||
@ -200,9 +200,9 @@ spec:
|
||||
{{- if .Values.global.seaweedfs.enableSecurity }}
|
||||
{{- if .Values.filer.s3.httpsPort }}
|
||||
-s3.port.https={{ .Values.filer.s3.httpsPort }} \
|
||||
{{- end }}
|
||||
{{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }}
|
||||
{{- end }}
|
||||
{{- end }}
|
||||
{{- if .Values.filer.s3.enableAuth }}
|
||||
-s3.config=/etc/sw/seaweedfs_s3_config \
|
||||
{{- end }}
|
||||
|
||||
@ -127,9 +127,9 @@ spec:
|
||||
{{- if .Values.global.seaweedfs.enableSecurity }}
|
||||
{{- if .Values.s3.httpsPort }}
|
||||
-port.https={{ .Values.s3.httpsPort }} \
|
||||
{{- end }}
|
||||
{{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "") | nindent 14 }}
|
||||
{{- end }}
|
||||
{{- end }}
|
||||
{{- if .Values.s3.domainName }}
|
||||
-domainName={{ .Values.s3.domainName }} \
|
||||
{{- end }}
|
||||
|
||||
@ -22,7 +22,7 @@ require (
|
||||
github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3 // indirect
|
||||
github.com/Azure/go-ntlmssp v0.1.0 // indirect
|
||||
github.com/Azure/go-ntlmssp v0.1.1 // indirect
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 // indirect
|
||||
github.com/FilenCloudDienste/filen-sdk-go v0.0.38 // indirect
|
||||
github.com/Files-com/files-sdk-go/v3 v3.2.264 // indirect
|
||||
|
||||
@ -51,8 +51,8 @@ github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4 h1:jWQK1GI+LeGGUKBAD
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4/go.mod h1:8mwH4klAm9DUgR2EEHyEEAQlRDvLPyg5fQry3y+cDew=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3 h1:sxgSqOB9CDToiaVFpxuvb5wGgGqWa3lCShcm5o0n3bE=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3/go.mod h1:XdED8i399lEVblYHTZM8eXaP07gv4Z58IL6ueMlVlrg=
|
||||
github.com/Azure/go-ntlmssp v0.1.0 h1:DjFo6YtWzNqNvQdrwEyr/e4nhU3vRiwenz5QX7sFz+A=
|
||||
github.com/Azure/go-ntlmssp v0.1.0/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk=
|
||||
github.com/Azure/go-ntlmssp v0.1.1 h1:l+FM/EEMb0U9QZE7mKNEDw5Mu3mFiaa2GKOoTSsNDPw=
|
||||
github.com/Azure/go-ntlmssp v0.1.1/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk=
|
||||
github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM=
|
||||
github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE=
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgvJqCH0sFfrBUTnUJSBrBf7++ypk+twtRs=
|
||||
|
||||
@ -23,6 +23,7 @@ type NfsOptions struct {
|
||||
readOnly *bool
|
||||
allowedClients *string
|
||||
volumeServerAccess *string
|
||||
portmapBind *string
|
||||
}
|
||||
|
||||
func init() {
|
||||
@ -34,6 +35,7 @@ func init() {
|
||||
nfsStandaloneOptions.readOnly = cmdNfs.Flag.Bool("readOnly", false, "export the filer path as read only")
|
||||
nfsStandaloneOptions.allowedClients = cmdNfs.Flag.String("allowedClients", "", "comma-separated client IPs, hostnames, or CIDRs allowed to connect")
|
||||
nfsStandaloneOptions.volumeServerAccess = cmdNfs.Flag.String("volumeServerAccess", "direct", "access volume servers by [direct|publicUrl|filerProxy]")
|
||||
nfsStandaloneOptions.portmapBind = cmdNfs.Flag.String("portmap.bind", "", "when set, bind a built-in portmap v2 responder on <ip>:111 so plain `mount -t nfs` works without client-side portmap bypass. Empty disables it. Binding port 111 requires root or CAP_NET_BIND_SERVICE and must not conflict with a system rpcbind.")
|
||||
}
|
||||
|
||||
var cmdNfs = &Command{
|
||||
@ -53,6 +55,26 @@ Safer defaults (since export ACLs are still not implemented):
|
||||
|
||||
Override -ip.bind to a routable address only after you have reviewed
|
||||
-allowedClients and the readiness of the rest of your deployment.
|
||||
|
||||
Mounting from a Linux client
|
||||
----------------------------
|
||||
The server does not run portmap/rpcbind by default. That means Linux
|
||||
mount.nfs, which queries portmap on port 111 first, will fail with
|
||||
"portmap query failed" against the plain form:
|
||||
|
||||
mount -t nfs -o nfsvers=3,nolock <host>:/export /mnt
|
||||
|
||||
Either tell the client to bypass portmap:
|
||||
|
||||
mount -t nfs -o nfsvers=3,nolock,port=2049,mountport=2049,\
|
||||
proto=tcp,mountproto=tcp <host>:/export /mnt
|
||||
|
||||
or enable the built-in portmap responder on the server:
|
||||
|
||||
weed nfs ... -portmap.bind=0.0.0.0
|
||||
|
||||
Binding port 111 requires root or CAP_NET_BIND_SERVICE and must not
|
||||
collide with a system rpcbind.
|
||||
`,
|
||||
}
|
||||
|
||||
@ -85,6 +107,7 @@ func runNfs(cmd *Command, args []string) bool {
|
||||
AllowedClients: util.StringSplit(*nfsStandaloneOptions.allowedClients, ","),
|
||||
VolumeServerAccess: *nfsStandaloneOptions.volumeServerAccess,
|
||||
GrpcDialOption: grpcDialOption,
|
||||
PortmapBind: *nfsStandaloneOptions.portmapBind,
|
||||
})
|
||||
if err != nil {
|
||||
glog.Errorf("NFS Server startup error: %v", err)
|
||||
|
||||
443
weed/server/nfs/portmap.go
Normal file
443
weed/server/nfs/portmap.go
Normal file
@ -0,0 +1,443 @@
|
||||
package nfs
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
)
|
||||
|
||||
// Minimal PORTMAP v2 responder.
|
||||
//
|
||||
// The upstream willscott/go-nfs library serves NFSv3 and MOUNT on a single TCP
|
||||
// port and deliberately does not register with portmap (RPC program 100000).
|
||||
// Linux mount.nfs, however, queries portmap on port 111 before sending the
|
||||
// MOUNT RPC, so the plain `mount -t nfs host:/export /mnt` command fails
|
||||
// against a default `weed nfs` deployment.
|
||||
//
|
||||
// When enabled, this responder binds the privileged port 111 (RFC 1833) on
|
||||
// both TCP and UDP and answers the subset of PORTMAP v2 calls that standard
|
||||
// Linux clients make: PMAP_NULL, PMAP_GETPORT and PMAP_DUMP. It refuses
|
||||
// registration from third parties (PMAP_SET / PMAP_UNSET return false) and
|
||||
// only exposes the programs that weed itself serves.
|
||||
//
|
||||
// References: RFC 1833 (Portmap v2), RFC 5531 (RPC).
|
||||
const (
|
||||
portmapProgram = 100000
|
||||
portmapVersion = 2
|
||||
portmapPort = 111
|
||||
|
||||
pmapProcNull = 0
|
||||
pmapProcSet = 1
|
||||
pmapProcUnset = 2
|
||||
pmapProcGetPort = 3
|
||||
pmapProcDump = 4
|
||||
|
||||
ipProtoTCP = 6
|
||||
ipProtoUDP = 17
|
||||
|
||||
nfsProgram = 100003
|
||||
mountProgram = 100005
|
||||
|
||||
// RPC
|
||||
rpcMsgCall = 0
|
||||
rpcMsgReply = 1
|
||||
|
||||
rpcMsgAccepted = 0
|
||||
|
||||
rpcAcceptSuccess = 0
|
||||
rpcAcceptProgUnavail = 1
|
||||
rpcAcceptProgMismatch = 2
|
||||
rpcAcceptProcUnavail = 3
|
||||
rpcAcceptGarbageArgs = 4
|
||||
|
||||
rpcAuthNone = 0
|
||||
|
||||
// Defensive limits. Portmap messages are tiny in practice; these caps
|
||||
// protect the responder from large or slow reads.
|
||||
portmapMaxRecord = 64 * 1024
|
||||
|
||||
// Per-connection read/write deadlines on the TCP listener. The idle
|
||||
// timeout bounds how long we wait for the next request on an otherwise
|
||||
// quiet connection; the IO timeout bounds a single read or write once
|
||||
// one is in flight. Both guard against slowloris-style stalls on the
|
||||
// privileged port 111.
|
||||
portmapTCPIdleTimeout = 30 * time.Second
|
||||
portmapTCPIOTimeout = 10 * time.Second
|
||||
|
||||
// Back-off applied before retrying after a non-fatal listener error
|
||||
// (e.g. EMFILE on TCP Accept, or a transient UDP read failure) so we
|
||||
// don't busy-loop when the host is under pressure.
|
||||
portmapRetryBackoff = 50 * time.Millisecond
|
||||
)
|
||||
|
||||
type portmapEntry struct {
|
||||
Program uint32
|
||||
Version uint32
|
||||
Protocol uint32
|
||||
Port uint32
|
||||
}
|
||||
|
||||
type portmapServer struct {
|
||||
bindIP string
|
||||
port int
|
||||
entries []portmapEntry
|
||||
|
||||
tcpListener net.Listener
|
||||
udpConn *net.UDPConn
|
||||
|
||||
// mu guards closed and conns. It is held only for bookkeeping, never
|
||||
// across network IO.
|
||||
mu sync.Mutex
|
||||
closed bool
|
||||
conns map[net.Conn]struct{}
|
||||
// done is closed exactly once by Close() so that background loops can
|
||||
// interrupt a retry-backoff sleep instead of waiting it out.
|
||||
done chan struct{}
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// newPortmapServer builds a responder advertising the NFS services the caller
|
||||
// runs on nfsTCPPort. We expose NFS v3 TCP and MOUNT v3 TCP only: the
|
||||
// underlying library does not handle UDP or older MOUNT versions, so it would
|
||||
// be misleading to advertise them.
|
||||
func newPortmapServer(bindIP string, port int, nfsTCPPort uint32) *portmapServer {
|
||||
if port <= 0 {
|
||||
port = portmapPort
|
||||
}
|
||||
return &portmapServer{
|
||||
bindIP: bindIP,
|
||||
port: port,
|
||||
done: make(chan struct{}),
|
||||
entries: []portmapEntry{
|
||||
{Program: nfsProgram, Version: 3, Protocol: ipProtoTCP, Port: nfsTCPPort},
|
||||
{Program: mountProgram, Version: 3, Protocol: ipProtoTCP, Port: nfsTCPPort},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (ps *portmapServer) Start() error {
|
||||
addr := net.JoinHostPort(ps.bindIP, fmt.Sprintf("%d", ps.port))
|
||||
|
||||
tcpLn, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("portmap tcp listen %s: %w", addr, err)
|
||||
}
|
||||
udpAddr, err := net.ResolveUDPAddr("udp", addr)
|
||||
if err != nil {
|
||||
_ = tcpLn.Close()
|
||||
return fmt.Errorf("portmap udp resolve %s: %w", addr, err)
|
||||
}
|
||||
udpConn, err := net.ListenUDP("udp", udpAddr)
|
||||
if err != nil {
|
||||
_ = tcpLn.Close()
|
||||
return fmt.Errorf("portmap udp listen %s: %w", addr, err)
|
||||
}
|
||||
ps.tcpListener = tcpLn
|
||||
ps.udpConn = udpConn
|
||||
|
||||
ps.wg.Add(2)
|
||||
go func() {
|
||||
defer ps.wg.Done()
|
||||
ps.serveTCP()
|
||||
}()
|
||||
go func() {
|
||||
defer ps.wg.Done()
|
||||
ps.serveUDP()
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ps *portmapServer) Close() error {
|
||||
ps.mu.Lock()
|
||||
if ps.closed {
|
||||
ps.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
ps.closed = true
|
||||
conns := ps.conns
|
||||
ps.conns = nil
|
||||
close(ps.done)
|
||||
ps.mu.Unlock()
|
||||
|
||||
var first error
|
||||
if ps.tcpListener != nil {
|
||||
if err := ps.tcpListener.Close(); err != nil {
|
||||
first = err
|
||||
}
|
||||
}
|
||||
if ps.udpConn != nil {
|
||||
if err := ps.udpConn.Close(); err != nil && first == nil {
|
||||
first = err
|
||||
}
|
||||
}
|
||||
// Evict in-flight TCP handlers so Close() does not block on idle
|
||||
// clients; their read goroutines will unwind on the closed conn.
|
||||
for c := range conns {
|
||||
_ = c.Close()
|
||||
}
|
||||
ps.wg.Wait()
|
||||
return first
|
||||
}
|
||||
|
||||
func (ps *portmapServer) isClosed() bool {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
return ps.closed
|
||||
}
|
||||
|
||||
// addConn registers c for shutdown eviction. It returns false (and the
|
||||
// caller must drop c) if the server has already started shutting down.
|
||||
func (ps *portmapServer) addConn(c net.Conn) bool {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if ps.closed {
|
||||
return false
|
||||
}
|
||||
if ps.conns == nil {
|
||||
ps.conns = make(map[net.Conn]struct{})
|
||||
}
|
||||
ps.conns[c] = struct{}{}
|
||||
return true
|
||||
}
|
||||
|
||||
func (ps *portmapServer) removeConn(c net.Conn) {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
delete(ps.conns, c)
|
||||
}
|
||||
|
||||
func (ps *portmapServer) serveTCP() {
|
||||
for {
|
||||
conn, err := ps.tcpListener.Accept()
|
||||
if err != nil {
|
||||
if ps.isClosed() {
|
||||
return
|
||||
}
|
||||
// Non-fatal (e.g. EMFILE, EINTR): log and back off rather
|
||||
// than tear the listener down on a transient resource blip.
|
||||
// Wake early if Close() fires during the sleep.
|
||||
glog.V(1).Infof("portmap tcp accept: %v", err)
|
||||
select {
|
||||
case <-ps.done:
|
||||
return
|
||||
case <-time.After(portmapRetryBackoff):
|
||||
continue
|
||||
}
|
||||
}
|
||||
if !ps.addConn(conn) {
|
||||
_ = conn.Close()
|
||||
continue
|
||||
}
|
||||
ps.wg.Add(1)
|
||||
go func(c net.Conn) {
|
||||
defer ps.wg.Done()
|
||||
defer ps.removeConn(c)
|
||||
ps.handleTCPConn(c)
|
||||
}(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (ps *portmapServer) handleTCPConn(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
hdr := make([]byte, 4)
|
||||
for {
|
||||
_ = conn.SetReadDeadline(time.Now().Add(portmapTCPIdleTimeout))
|
||||
if _, err := io.ReadFull(conn, hdr); err != nil {
|
||||
return
|
||||
}
|
||||
mark := binary.BigEndian.Uint32(hdr)
|
||||
// Bit 31: last-fragment flag. Portmap messages are always single
|
||||
// fragment in practice; drop the connection if we see otherwise.
|
||||
if mark&(1<<31) == 0 {
|
||||
return
|
||||
}
|
||||
recLen := mark &^ (1 << 31)
|
||||
if recLen == 0 || recLen > portmapMaxRecord {
|
||||
return
|
||||
}
|
||||
buf := make([]byte, recLen)
|
||||
_ = conn.SetReadDeadline(time.Now().Add(portmapTCPIOTimeout))
|
||||
if _, err := io.ReadFull(conn, buf); err != nil {
|
||||
return
|
||||
}
|
||||
reply := ps.handleCall(buf)
|
||||
if reply == nil {
|
||||
continue
|
||||
}
|
||||
out := make([]byte, 4+len(reply))
|
||||
binary.BigEndian.PutUint32(out[0:4], uint32(len(reply))|(1<<31))
|
||||
copy(out[4:], reply)
|
||||
_ = conn.SetWriteDeadline(time.Now().Add(portmapTCPIOTimeout))
|
||||
if _, err := conn.Write(out); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ps *portmapServer) serveUDP() {
|
||||
buf := make([]byte, portmapMaxRecord)
|
||||
for {
|
||||
n, addr, err := ps.udpConn.ReadFromUDP(buf)
|
||||
if err != nil {
|
||||
if ps.isClosed() {
|
||||
return
|
||||
}
|
||||
// Transient read failure: log, back off, and keep the
|
||||
// responder alive instead of taking UDP portmap down.
|
||||
// Wake early if Close() fires during the sleep.
|
||||
glog.V(1).Infof("portmap udp read: %v", err)
|
||||
select {
|
||||
case <-ps.done:
|
||||
return
|
||||
case <-time.After(portmapRetryBackoff):
|
||||
continue
|
||||
}
|
||||
}
|
||||
reply := ps.handleCall(buf[:n])
|
||||
if reply == nil {
|
||||
continue
|
||||
}
|
||||
if _, err := ps.udpConn.WriteToUDP(reply, addr); err != nil {
|
||||
glog.V(1).Infof("portmap udp write to %s: %v", addr, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// handleCall parses one RPC CALL message and returns the encoded reply, or nil
|
||||
// if the call is malformed enough that we should drop it silently.
|
||||
func (ps *portmapServer) handleCall(callBuf []byte) []byte {
|
||||
xid, prog, vers, proc, args, err := parseRPCCall(callBuf)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
if prog != portmapProgram {
|
||||
return encodeAcceptedReply(xid, rpcAcceptProgUnavail, nil)
|
||||
}
|
||||
if vers != portmapVersion {
|
||||
// Program-version mismatch: RFC 5531 says we should return the
|
||||
// accepted range; keep it simple and report 2..2.
|
||||
body := make([]byte, 8)
|
||||
binary.BigEndian.PutUint32(body[0:4], portmapVersion)
|
||||
binary.BigEndian.PutUint32(body[4:8], portmapVersion)
|
||||
return encodeAcceptedReply(xid, rpcAcceptProgMismatch, body)
|
||||
}
|
||||
switch proc {
|
||||
case pmapProcNull:
|
||||
return encodeAcceptedReply(xid, rpcAcceptSuccess, nil)
|
||||
case pmapProcGetPort:
|
||||
if len(args) < 16 {
|
||||
return encodeAcceptedReply(xid, rpcAcceptGarbageArgs, nil)
|
||||
}
|
||||
q := portmapEntry{
|
||||
Program: binary.BigEndian.Uint32(args[0:4]),
|
||||
Version: binary.BigEndian.Uint32(args[4:8]),
|
||||
Protocol: binary.BigEndian.Uint32(args[8:12]),
|
||||
}
|
||||
port := uint32(0)
|
||||
for _, e := range ps.entries {
|
||||
if e.Program == q.Program && e.Version == q.Version && e.Protocol == q.Protocol {
|
||||
port = e.Port
|
||||
break
|
||||
}
|
||||
}
|
||||
body := make([]byte, 4)
|
||||
binary.BigEndian.PutUint32(body, port)
|
||||
return encodeAcceptedReply(xid, rpcAcceptSuccess, body)
|
||||
case pmapProcDump:
|
||||
// Each entry is 4-byte value_follows + 16-byte mapping = 20 bytes,
|
||||
// plus a 4-byte terminator value_follows=FALSE.
|
||||
body := make([]byte, 0, 20*len(ps.entries)+4)
|
||||
for _, e := range ps.entries {
|
||||
chunk := make([]byte, 20)
|
||||
binary.BigEndian.PutUint32(chunk[0:4], 1) // value_follows = TRUE
|
||||
binary.BigEndian.PutUint32(chunk[4:8], e.Program)
|
||||
binary.BigEndian.PutUint32(chunk[8:12], e.Version)
|
||||
binary.BigEndian.PutUint32(chunk[12:16], e.Protocol)
|
||||
binary.BigEndian.PutUint32(chunk[16:20], e.Port)
|
||||
body = append(body, chunk...)
|
||||
}
|
||||
end := make([]byte, 4) // value_follows = FALSE
|
||||
body = append(body, end...)
|
||||
return encodeAcceptedReply(xid, rpcAcceptSuccess, body)
|
||||
case pmapProcSet, pmapProcUnset:
|
||||
// Don't accept third-party registrations. bool=FALSE.
|
||||
body := make([]byte, 4)
|
||||
return encodeAcceptedReply(xid, rpcAcceptSuccess, body)
|
||||
default:
|
||||
return encodeAcceptedReply(xid, rpcAcceptProcUnavail, nil)
|
||||
}
|
||||
}
|
||||
|
||||
// parseRPCCall parses the fixed portion of an RPC CALL header and returns the
|
||||
// remaining procedure arguments. It skips both opaque_auth fields (cred and
|
||||
// verf) so callers get a buffer starting at the procedure arguments.
|
||||
func parseRPCCall(buf []byte) (xid, prog, vers, proc uint32, args []byte, err error) {
|
||||
// Minimum header: xid + msg_type + rpcvers + prog + vers + proc + 2x
|
||||
// (flavor + len) = 6*4 + 2*8 = 40 bytes.
|
||||
const minHeader = 40
|
||||
if len(buf) < minHeader {
|
||||
err = fmt.Errorf("rpc call too short: %d bytes", len(buf))
|
||||
return
|
||||
}
|
||||
xid = binary.BigEndian.Uint32(buf[0:4])
|
||||
if msgType := binary.BigEndian.Uint32(buf[4:8]); msgType != rpcMsgCall {
|
||||
err = fmt.Errorf("not an rpc call: msg_type=%d", msgType)
|
||||
return
|
||||
}
|
||||
if rpcvers := binary.BigEndian.Uint32(buf[8:12]); rpcvers != 2 {
|
||||
err = fmt.Errorf("unsupported rpc version %d", rpcvers)
|
||||
return
|
||||
}
|
||||
prog = binary.BigEndian.Uint32(buf[12:16])
|
||||
vers = binary.BigEndian.Uint32(buf[16:20])
|
||||
proc = binary.BigEndian.Uint32(buf[20:24])
|
||||
|
||||
p := 24
|
||||
for i := 0; i < 2; i++ {
|
||||
if len(buf) < p+8 {
|
||||
err = fmt.Errorf("truncated opaque_auth at offset %d", p)
|
||||
return
|
||||
}
|
||||
authLen := binary.BigEndian.Uint32(buf[p+4 : p+8])
|
||||
// Validate before applying the XDR 4-byte padding so that
|
||||
// lengths near uint32 max can't wrap to a tiny padded value.
|
||||
if authLen > uint32(portmapMaxRecord) {
|
||||
err = errors.New("opaque_auth length exceeds limit")
|
||||
return
|
||||
}
|
||||
padded := (authLen + 3) &^ 3
|
||||
end := uint64(p) + 8 + uint64(padded)
|
||||
if end > uint64(len(buf)) {
|
||||
err = fmt.Errorf("truncated opaque_auth body at offset %d (len=%d)", p, authLen)
|
||||
return
|
||||
}
|
||||
p = int(end)
|
||||
}
|
||||
args = buf[p:]
|
||||
return
|
||||
}
|
||||
|
||||
// encodeAcceptedReply builds a MSG_ACCEPTED reply with the given accept_stat.
|
||||
// body is the already-XDR-encoded data that follows accept_stat in the reply.
|
||||
// For SUCCESS it is the procedure result; it is nil for most error
|
||||
// accept_stat values (PROG_UNAVAIL, PROC_UNAVAIL, GARBAGE_ARGS) but is
|
||||
// non-nil for PROG_MISMATCH, which carries a struct { uint32 low; uint32
|
||||
// high; } mismatch_info range per RFC 5531 §9.
|
||||
func encodeAcceptedReply(xid, acceptStat uint32, body []byte) []byte {
|
||||
out := make([]byte, 24+len(body))
|
||||
binary.BigEndian.PutUint32(out[0:4], xid)
|
||||
binary.BigEndian.PutUint32(out[4:8], rpcMsgReply)
|
||||
binary.BigEndian.PutUint32(out[8:12], rpcMsgAccepted)
|
||||
// verf: AUTH_NONE, zero-length opaque
|
||||
binary.BigEndian.PutUint32(out[12:16], rpcAuthNone)
|
||||
binary.BigEndian.PutUint32(out[16:20], 0)
|
||||
binary.BigEndian.PutUint32(out[20:24], acceptStat)
|
||||
copy(out[24:], body)
|
||||
return out
|
||||
}
|
||||
416
weed/server/nfs/portmap_test.go
Normal file
416
weed/server/nfs/portmap_test.go
Normal file
@ -0,0 +1,416 @@
|
||||
package nfs
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"io"
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func buildRPCCall(t *testing.T, xid, prog, vers, proc uint32, credBody, verfBody, args []byte) []byte {
|
||||
t.Helper()
|
||||
pad := func(b []byte) []byte {
|
||||
r := len(b) % 4
|
||||
if r == 0 {
|
||||
return b
|
||||
}
|
||||
out := make([]byte, len(b)+4-r)
|
||||
copy(out, b)
|
||||
return out
|
||||
}
|
||||
buf := new(bytes.Buffer)
|
||||
write := func(v uint32) {
|
||||
var b [4]byte
|
||||
binary.BigEndian.PutUint32(b[:], v)
|
||||
buf.Write(b[:])
|
||||
}
|
||||
write(xid)
|
||||
write(rpcMsgCall)
|
||||
write(2) // rpcvers
|
||||
write(prog)
|
||||
write(vers)
|
||||
write(proc)
|
||||
// cred
|
||||
write(rpcAuthNone)
|
||||
write(uint32(len(credBody)))
|
||||
buf.Write(pad(credBody))
|
||||
// verf
|
||||
write(rpcAuthNone)
|
||||
write(uint32(len(verfBody)))
|
||||
buf.Write(pad(verfBody))
|
||||
buf.Write(args)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func parseAcceptedReply(t *testing.T, reply []byte) (xid, acceptStat uint32, body []byte) {
|
||||
t.Helper()
|
||||
if len(reply) < 24 {
|
||||
t.Fatalf("reply too short: %d bytes", len(reply))
|
||||
}
|
||||
xid = binary.BigEndian.Uint32(reply[0:4])
|
||||
if mt := binary.BigEndian.Uint32(reply[4:8]); mt != rpcMsgReply {
|
||||
t.Fatalf("msg_type=%d, want REPLY", mt)
|
||||
}
|
||||
if rs := binary.BigEndian.Uint32(reply[8:12]); rs != rpcMsgAccepted {
|
||||
t.Fatalf("reply_stat=%d, want ACCEPTED", rs)
|
||||
}
|
||||
// verf
|
||||
verfLen := binary.BigEndian.Uint32(reply[16:20])
|
||||
if verfLen != 0 {
|
||||
t.Fatalf("unexpected verf length %d", verfLen)
|
||||
}
|
||||
acceptStat = binary.BigEndian.Uint32(reply[20:24])
|
||||
body = reply[24:]
|
||||
return
|
||||
}
|
||||
|
||||
func newTestPortmap() *portmapServer {
|
||||
return newPortmapServer("127.0.0.1", portmapPort, 2049)
|
||||
}
|
||||
|
||||
func TestParseRPCCall_SkipsAuth(t *testing.T) {
|
||||
cred := []byte("hello") // 5 bytes -> padded to 8
|
||||
verf := []byte{}
|
||||
args := []byte{0x01, 0x02, 0x03, 0x04}
|
||||
msg := buildRPCCall(t, 42, portmapProgram, portmapVersion, pmapProcNull, cred, verf, args)
|
||||
|
||||
xid, prog, vers, proc, gotArgs, err := parseRPCCall(msg)
|
||||
if err != nil {
|
||||
t.Fatalf("parseRPCCall: %v", err)
|
||||
}
|
||||
if xid != 42 || prog != portmapProgram || vers != portmapVersion || proc != pmapProcNull {
|
||||
t.Fatalf("header mismatch: xid=%d prog=%d vers=%d proc=%d", xid, prog, vers, proc)
|
||||
}
|
||||
if !bytes.Equal(gotArgs, args) {
|
||||
t.Fatalf("args mismatch: got %x want %x", gotArgs, args)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseRPCCall_RejectsReply(t *testing.T) {
|
||||
buf := make([]byte, 40)
|
||||
binary.BigEndian.PutUint32(buf[4:8], rpcMsgReply)
|
||||
if _, _, _, _, _, err := parseRPCCall(buf); err == nil {
|
||||
t.Fatal("expected error on reply-typed message")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseRPCCall_TruncatedAuth(t *testing.T) {
|
||||
// Claim huge cred length but provide no body.
|
||||
buf := make([]byte, 40)
|
||||
binary.BigEndian.PutUint32(buf[4:8], rpcMsgCall)
|
||||
binary.BigEndian.PutUint32(buf[8:12], 2)
|
||||
binary.BigEndian.PutUint32(buf[28:32], 1000) // cred len
|
||||
if _, _, _, _, _, err := parseRPCCall(buf); err == nil {
|
||||
t.Fatal("expected error on truncated auth")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_Null(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
msg := buildRPCCall(t, 7, portmapProgram, portmapVersion, pmapProcNull, nil, nil, nil)
|
||||
reply := ps.handleCall(msg)
|
||||
xid, acc, body := parseAcceptedReply(t, reply)
|
||||
if xid != 7 || acc != rpcAcceptSuccess || len(body) != 0 {
|
||||
t.Fatalf("null reply xid=%d acc=%d body=%x", xid, acc, body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_GetPort_HitAndMiss(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
|
||||
buildQuery := func(prog, vers, prot uint32) []byte {
|
||||
args := make([]byte, 16)
|
||||
binary.BigEndian.PutUint32(args[0:4], prog)
|
||||
binary.BigEndian.PutUint32(args[4:8], vers)
|
||||
binary.BigEndian.PutUint32(args[8:12], prot)
|
||||
// port field is ignored by the server; leave zero
|
||||
return args
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
prog, vers, prot uint32
|
||||
wantPort uint32
|
||||
}{
|
||||
{"nfs-v3-tcp-hit", nfsProgram, 3, ipProtoTCP, 2049},
|
||||
{"mount-v3-tcp-hit", mountProgram, 3, ipProtoTCP, 2049},
|
||||
{"mount-v1-tcp-miss", mountProgram, 1, ipProtoTCP, 0},
|
||||
{"nfs-v3-udp-miss", nfsProgram, 3, ipProtoUDP, 0},
|
||||
{"nlm-miss", 100021, 4, ipProtoTCP, 0},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
msg := buildRPCCall(t, 11, portmapProgram, portmapVersion, pmapProcGetPort, nil, nil, buildQuery(tc.prog, tc.vers, tc.prot))
|
||||
reply := ps.handleCall(msg)
|
||||
xid, acc, body := parseAcceptedReply(t, reply)
|
||||
if xid != 11 {
|
||||
t.Fatalf("xid=%d want 11", xid)
|
||||
}
|
||||
if acc != rpcAcceptSuccess {
|
||||
t.Fatalf("acc=%d want SUCCESS", acc)
|
||||
}
|
||||
if len(body) != 4 {
|
||||
t.Fatalf("getport body len=%d want 4", len(body))
|
||||
}
|
||||
got := binary.BigEndian.Uint32(body)
|
||||
if got != tc.wantPort {
|
||||
t.Fatalf("port=%d want %d", got, tc.wantPort)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_Dump(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
msg := buildRPCCall(t, 13, portmapProgram, portmapVersion, pmapProcDump, nil, nil, nil)
|
||||
reply := ps.handleCall(msg)
|
||||
_, acc, body := parseAcceptedReply(t, reply)
|
||||
if acc != rpcAcceptSuccess {
|
||||
t.Fatalf("acc=%d", acc)
|
||||
}
|
||||
var entries []portmapEntry
|
||||
p := 0
|
||||
for p+4 <= len(body) {
|
||||
vf := binary.BigEndian.Uint32(body[p : p+4])
|
||||
p += 4
|
||||
if vf == 0 {
|
||||
break
|
||||
}
|
||||
if p+16 > len(body) {
|
||||
t.Fatalf("truncated entry at %d", p)
|
||||
}
|
||||
entries = append(entries, portmapEntry{
|
||||
Program: binary.BigEndian.Uint32(body[p : p+4]),
|
||||
Version: binary.BigEndian.Uint32(body[p+4 : p+8]),
|
||||
Protocol: binary.BigEndian.Uint32(body[p+8 : p+12]),
|
||||
Port: binary.BigEndian.Uint32(body[p+12 : p+16]),
|
||||
})
|
||||
p += 16
|
||||
}
|
||||
if len(entries) != 2 {
|
||||
t.Fatalf("got %d dump entries, want 2: %+v", len(entries), entries)
|
||||
}
|
||||
wantSet := map[portmapEntry]bool{
|
||||
{Program: nfsProgram, Version: 3, Protocol: ipProtoTCP, Port: 2049}: false,
|
||||
{Program: mountProgram, Version: 3, Protocol: ipProtoTCP, Port: 2049}: false,
|
||||
}
|
||||
for _, e := range entries {
|
||||
if _, ok := wantSet[e]; !ok {
|
||||
t.Fatalf("unexpected dump entry %+v", e)
|
||||
}
|
||||
wantSet[e] = true
|
||||
}
|
||||
for e, seen := range wantSet {
|
||||
if !seen {
|
||||
t.Fatalf("missing dump entry %+v", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_UnknownProg(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
msg := buildRPCCall(t, 1, 999999, 1, 0, nil, nil, nil)
|
||||
reply := ps.handleCall(msg)
|
||||
_, acc, _ := parseAcceptedReply(t, reply)
|
||||
if acc != rpcAcceptProgUnavail {
|
||||
t.Fatalf("acc=%d want PROG_UNAVAIL", acc)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_VersionMismatch(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
msg := buildRPCCall(t, 1, portmapProgram, 42, pmapProcNull, nil, nil, nil)
|
||||
reply := ps.handleCall(msg)
|
||||
_, acc, body := parseAcceptedReply(t, reply)
|
||||
if acc != rpcAcceptProgMismatch {
|
||||
t.Fatalf("acc=%d want PROG_MISMATCH", acc)
|
||||
}
|
||||
if len(body) != 8 {
|
||||
t.Fatalf("mismatch body len=%d want 8", len(body))
|
||||
}
|
||||
lo := binary.BigEndian.Uint32(body[0:4])
|
||||
hi := binary.BigEndian.Uint32(body[4:8])
|
||||
if lo != portmapVersion || hi != portmapVersion {
|
||||
t.Fatalf("mismatch range lo=%d hi=%d", lo, hi)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_UnknownProc(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
msg := buildRPCCall(t, 1, portmapProgram, portmapVersion, 42, nil, nil, nil)
|
||||
reply := ps.handleCall(msg)
|
||||
_, acc, _ := parseAcceptedReply(t, reply)
|
||||
if acc != rpcAcceptProcUnavail {
|
||||
t.Fatalf("acc=%d want PROC_UNAVAIL", acc)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCall_SetRefused(t *testing.T) {
|
||||
ps := newTestPortmap()
|
||||
args := make([]byte, 16) // mapping struct
|
||||
msg := buildRPCCall(t, 1, portmapProgram, portmapVersion, pmapProcSet, nil, nil, args)
|
||||
reply := ps.handleCall(msg)
|
||||
_, acc, body := parseAcceptedReply(t, reply)
|
||||
if acc != rpcAcceptSuccess {
|
||||
t.Fatalf("acc=%d", acc)
|
||||
}
|
||||
if len(body) != 4 || binary.BigEndian.Uint32(body) != 0 {
|
||||
t.Fatalf("PMAP_SET must return FALSE, got %x", body)
|
||||
}
|
||||
}
|
||||
|
||||
// pickFreePort asks the OS for an unused high port by opening and closing a
|
||||
// listener on it. Used so the end-to-end tests can run in parallel without
|
||||
// stepping on the privileged default port 111.
|
||||
func pickFreePort(t *testing.T) int {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
defer ln.Close()
|
||||
return ln.Addr().(*net.TCPAddr).Port
|
||||
}
|
||||
|
||||
func TestPortmapServer_UDPGetPort(t *testing.T) {
|
||||
port := pickFreePort(t)
|
||||
ps := newPortmapServer("127.0.0.1", port, 2049)
|
||||
if err := ps.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = ps.Close() })
|
||||
|
||||
args := make([]byte, 16)
|
||||
binary.BigEndian.PutUint32(args[0:4], nfsProgram)
|
||||
binary.BigEndian.PutUint32(args[4:8], 3)
|
||||
binary.BigEndian.PutUint32(args[8:12], ipProtoTCP)
|
||||
msg := buildRPCCall(t, 99, portmapProgram, portmapVersion, pmapProcGetPort, nil, nil, args)
|
||||
|
||||
conn, err := net.Dial("udp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)))
|
||||
if err != nil {
|
||||
t.Fatalf("dial udp: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
_ = conn.SetDeadline(time.Now().Add(2 * time.Second))
|
||||
if _, err := conn.Write(msg); err != nil {
|
||||
t.Fatalf("write: %v", err)
|
||||
}
|
||||
buf := make([]byte, 4096)
|
||||
n, err := conn.Read(buf)
|
||||
if err != nil {
|
||||
t.Fatalf("read: %v", err)
|
||||
}
|
||||
xid, acc, body := parseAcceptedReply(t, buf[:n])
|
||||
if xid != 99 || acc != rpcAcceptSuccess || len(body) != 4 {
|
||||
t.Fatalf("bad reply xid=%d acc=%d body=%x", xid, acc, body)
|
||||
}
|
||||
if got := binary.BigEndian.Uint32(body); got != 2049 {
|
||||
t.Fatalf("udp getport port=%d want 2049", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPortmapServer_CloseEvictsIdleTCPConn(t *testing.T) {
|
||||
port := pickFreePort(t)
|
||||
ps := newPortmapServer("127.0.0.1", port, 2049)
|
||||
if err := ps.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
|
||||
conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)))
|
||||
if err != nil {
|
||||
_ = ps.Close()
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
// Issue one call and read its reply so the server-side connection is
|
||||
// definitely registered before we trigger shutdown.
|
||||
msg := buildRPCCall(t, 1, portmapProgram, portmapVersion, pmapProcNull, nil, nil, nil)
|
||||
var mark [4]byte
|
||||
binary.BigEndian.PutUint32(mark[:], uint32(len(msg))|(1<<31))
|
||||
_ = conn.SetDeadline(time.Now().Add(2 * time.Second))
|
||||
if _, err := conn.Write(mark[:]); err != nil {
|
||||
t.Fatalf("write mark: %v", err)
|
||||
}
|
||||
if _, err := conn.Write(msg); err != nil {
|
||||
t.Fatalf("write msg: %v", err)
|
||||
}
|
||||
if _, err := io.ReadFull(conn, mark[:]); err != nil {
|
||||
t.Fatalf("read mark: %v", err)
|
||||
}
|
||||
rlen := binary.BigEndian.Uint32(mark[:]) &^ (1 << 31)
|
||||
if _, err := io.ReadFull(conn, make([]byte, rlen)); err != nil {
|
||||
t.Fatalf("read body: %v", err)
|
||||
}
|
||||
|
||||
// Close must return long before the TCP idle deadline (30s) — in
|
||||
// other words, the server must actively close the idle conn rather
|
||||
// than wait for the deadline or for the client to disconnect.
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- ps.Close() }()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Close did not return within 2s; in-flight conn not evicted")
|
||||
}
|
||||
|
||||
_ = conn.SetReadDeadline(time.Now().Add(1 * time.Second))
|
||||
if _, err := conn.Read(make([]byte, 4)); err == nil {
|
||||
t.Fatal("expected read error on client conn after server Close")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPortmapServer_TCPGetPort(t *testing.T) {
|
||||
port := pickFreePort(t)
|
||||
ps := newPortmapServer("127.0.0.1", port, 2049)
|
||||
if err := ps.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = ps.Close() })
|
||||
|
||||
args := make([]byte, 16)
|
||||
binary.BigEndian.PutUint32(args[0:4], mountProgram)
|
||||
binary.BigEndian.PutUint32(args[4:8], 3)
|
||||
binary.BigEndian.PutUint32(args[8:12], ipProtoTCP)
|
||||
msg := buildRPCCall(t, 123, portmapProgram, portmapVersion, pmapProcGetPort, nil, nil, args)
|
||||
|
||||
conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)))
|
||||
if err != nil {
|
||||
t.Fatalf("dial tcp: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
_ = conn.SetDeadline(time.Now().Add(2 * time.Second))
|
||||
|
||||
// record mark: last-fragment bit + length
|
||||
var mark [4]byte
|
||||
binary.BigEndian.PutUint32(mark[:], uint32(len(msg))|(1<<31))
|
||||
if _, err := conn.Write(mark[:]); err != nil {
|
||||
t.Fatalf("write mark: %v", err)
|
||||
}
|
||||
if _, err := conn.Write(msg); err != nil {
|
||||
t.Fatalf("write msg: %v", err)
|
||||
}
|
||||
|
||||
var rmark [4]byte
|
||||
if _, err := io.ReadFull(conn, rmark[:]); err != nil {
|
||||
t.Fatalf("read mark: %v", err)
|
||||
}
|
||||
rlen := binary.BigEndian.Uint32(rmark[:]) &^ (1 << 31)
|
||||
buf := make([]byte, rlen)
|
||||
if _, err := io.ReadFull(conn, buf); err != nil {
|
||||
t.Fatalf("read body: %v", err)
|
||||
}
|
||||
xid, acc, body := parseAcceptedReply(t, buf)
|
||||
if xid != 123 || acc != rpcAcceptSuccess || len(body) != 4 {
|
||||
t.Fatalf("bad reply xid=%d acc=%d body=%x", xid, acc, body)
|
||||
}
|
||||
if got := binary.BigEndian.Uint32(body); got != 2049 {
|
||||
t.Fatalf("tcp getport port=%d want 2049", got)
|
||||
}
|
||||
}
|
||||
@ -26,6 +26,11 @@ type Option struct {
|
||||
AllowedClients []string
|
||||
VolumeServerAccess string
|
||||
GrpcDialOption grpc.DialOption
|
||||
// PortmapBind, when non-empty, enables a built-in portmap v2 responder
|
||||
// on <PortmapBind>:111 advertising the NFS v3 and MOUNT v3 services at
|
||||
// Port. Empty (the default) disables portmap; clients must then bypass
|
||||
// portmap with mount -o port=,mountport=,proto=tcp,mountproto=tcp.
|
||||
PortmapBind string
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
@ -93,9 +98,41 @@ func (s *Server) Start() error {
|
||||
return fmt.Errorf("listen nfs on %s:%d: %w", s.option.BindIp, s.option.Port, err)
|
||||
}
|
||||
|
||||
var portmap *portmapServer
|
||||
if s.option.PortmapBind != "" {
|
||||
portmap = newPortmapServer(s.option.PortmapBind, portmapPort, uint32(s.option.Port))
|
||||
if pmErr := portmap.Start(); pmErr != nil {
|
||||
_ = listener.Close()
|
||||
return fmt.Errorf("start portmap: %w", pmErr)
|
||||
}
|
||||
glog.V(0).Infof("NFS portmap responder listening on %s:%d (NFS v3 tcp=%d, MOUNT v3 tcp=%d)",
|
||||
s.option.PortmapBind, portmapPort, s.option.Port, s.option.Port)
|
||||
defer func() {
|
||||
if portmap != nil {
|
||||
_ = portmap.Close()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
s.logMountHint()
|
||||
return s.serve(listener)
|
||||
}
|
||||
|
||||
// logMountHint prints a copy-pasteable Linux mount command so operators can
|
||||
// see at startup how to mount the export from a client. The go-nfs library
|
||||
// does not run portmap, so without -portmap.bind the client must bypass
|
||||
// portmap via -o port=,mountport=,proto=tcp,mountproto=tcp.
|
||||
func (s *Server) logMountHint() {
|
||||
exportPath := string(s.exportRoot)
|
||||
if s.option.PortmapBind != "" {
|
||||
glog.V(0).Infof("mount example: mount -t nfs -o nfsvers=3,nolock <host>:%s <mountpoint>", exportPath)
|
||||
return
|
||||
}
|
||||
glog.V(0).Infof("mount example (bypasses portmap): mount -t nfs -o nfsvers=3,nolock,noacl,port=%d,mountport=%d,proto=tcp,mountproto=tcp <host>:%s <mountpoint>",
|
||||
s.option.Port, s.option.Port, exportPath)
|
||||
glog.V(0).Infof("tip: pass -portmap.bind to enable the built-in portmap responder on port 111 so plain `mount -t nfs host:%s /mnt` works.", exportPath)
|
||||
}
|
||||
|
||||
func (s *Server) serve(listener net.Listener) error {
|
||||
if s.filerClient != nil {
|
||||
defer s.filerClient.Close()
|
||||
|
||||
@ -807,7 +807,7 @@ func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) {
|
||||
}
|
||||
volCount := diskLocation.VolumesLen()
|
||||
ecShardCount := diskLocation.EcShardCount()
|
||||
maxVolumeCount := int32(volCount) + int32((ecShardCount+erasure_coding.DataShardsCount)/erasure_coding.DataShardsCount)
|
||||
maxVolumeCount := int32(volCount) + int32((ecShardCount+erasure_coding.DataShardsCount-1)/erasure_coding.DataShardsCount)
|
||||
if unclaimedSpaces > int64(volumeSizeLimit) {
|
||||
maxVolumeCount += int32(uint64(unclaimedSpaces)/volumeSizeLimit) - 1
|
||||
}
|
||||
|
||||
@ -23,6 +23,12 @@ type Disk struct {
|
||||
ecShardsLock sync.RWMutex
|
||||
}
|
||||
|
||||
// ecShardSlots returns the number of volume slots consumed by the given
|
||||
// number of EC shards, rounded up to whole-volume equivalents.
|
||||
func ecShardSlots(ecShardCount int64) int64 {
|
||||
return (ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount
|
||||
}
|
||||
|
||||
func NewDisk(diskType string) *Disk {
|
||||
s := &Disk{}
|
||||
s.id = NodeId(diskType)
|
||||
@ -67,7 +73,7 @@ func (d *DiskUsages) ToDiskInfo() map[string]*master_pb.DiskInfo {
|
||||
m := &master_pb.DiskInfo{
|
||||
VolumeCount: diskUsageCounts.volumeCount,
|
||||
MaxVolumeCount: diskUsageCounts.maxVolumeCount,
|
||||
FreeVolumeCount: diskUsageCounts.maxVolumeCount - (diskUsageCounts.volumeCount - diskUsageCounts.remoteVolumeCount) - (diskUsageCounts.ecShardCount+1)/erasure_coding.DataShardsCount,
|
||||
FreeVolumeCount: diskUsageCounts.maxVolumeCount - (diskUsageCounts.volumeCount - diskUsageCounts.remoteVolumeCount) - ecShardSlots(diskUsageCounts.ecShardCount),
|
||||
ActiveVolumeCount: diskUsageCounts.activeVolumeCount,
|
||||
RemoteVolumeCount: diskUsageCounts.remoteVolumeCount,
|
||||
}
|
||||
@ -111,11 +117,7 @@ func (a *DiskUsageCounts) addDiskUsageCounts(b *DiskUsageCounts) {
|
||||
}
|
||||
|
||||
func (a *DiskUsageCounts) FreeSpace() int64 {
|
||||
freeVolumeSlotCount := a.maxVolumeCount + a.remoteVolumeCount - a.volumeCount
|
||||
if a.ecShardCount > 0 {
|
||||
freeVolumeSlotCount = freeVolumeSlotCount - a.ecShardCount/erasure_coding.DataShardsCount - 1
|
||||
}
|
||||
return freeVolumeSlotCount
|
||||
return a.maxVolumeCount + a.remoteVolumeCount - a.volumeCount - ecShardSlots(a.ecShardCount)
|
||||
}
|
||||
|
||||
func (du *DiskUsages) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
|
||||
@ -265,7 +267,7 @@ func (d *Disk) ToDiskInfo() *master_pb.DiskInfo {
|
||||
Type: string(d.Id()),
|
||||
VolumeCount: diskUsage.volumeCount,
|
||||
MaxVolumeCount: diskUsage.maxVolumeCount,
|
||||
FreeVolumeCount: diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - (diskUsage.ecShardCount+1)/erasure_coding.DataShardsCount,
|
||||
FreeVolumeCount: diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - ecShardSlots(diskUsage.ecShardCount),
|
||||
ActiveVolumeCount: diskUsage.activeVolumeCount,
|
||||
RemoteVolumeCount: diskUsage.remoteVolumeCount,
|
||||
DiskId: diskId,
|
||||
|
||||
@ -11,7 +11,6 @@ import (
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/stats"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
||||
)
|
||||
@ -272,10 +271,7 @@ func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
|
||||
func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 {
|
||||
t := n.getOrCreateDisk(option.DiskType)
|
||||
freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount)
|
||||
ecShardCount := atomic.LoadInt64(&t.ecShardCount)
|
||||
if ecShardCount > 0 {
|
||||
freeVolumeSlotCount = freeVolumeSlotCount - ecShardCount/erasure_coding.DataShardsCount - 1
|
||||
}
|
||||
freeVolumeSlotCount -= ecShardSlots(atomic.LoadInt64(&t.ecShardCount))
|
||||
return freeVolumeSlotCount
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
