champa-client/main.go

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "flag"
   7         "fmt"
   8         "io"
   9         "log"
  10         "net"
  11         "net/http"
  12         "net/url"
  13         "os"
  14         "sync"
  15         "time"
  16
  17         "github.com/xtaci/kcp-go/v5"
  18         "github.com/xtaci/smux"
  19         "www.bamsoftware.com/git/champa.git/noise"
  20         "www.bamsoftware.com/git/champa.git/turbotunnel"
  21 )
  22
  23 // smux streams will be closed after this much time without receiving data.
  24 const idleTimeout = 2 * time.Minute
  25
  26 // readKeyFromFile reads a key from a named file.
  27 func readKeyFromFile(filename string) ([]byte, error) {
  28         f, err := os.Open(filename)
  29         if err != nil {
  30                 return nil, err
  31         }
  32         defer f.Close()
  33         return noise.ReadKey(f)
  34 }
  35
  36 // noisePacketConn implements the net.PacketConn interface. It acts as an
  37 // intermediary between an upper layer and an inner net.PacketConn, decrypting
  38 // packets on ReadFrom and encrypting them on WriteTo.
  39 type noisePacketConn struct {
  40         sess *noise.Session
  41         net.PacketConn
  42 }
  43
  44 // readNoiseMessageOfTypeFrom returns the first complete Noise message whose
  45 // msgTime is wantedType, discarding messages of any other msgType.
  46 func readNoiseMessageOfTypeFrom(conn net.PacketConn, wantedType byte) ([]byte, net.Addr, error) {
  47         for {
  48                 msgType, msg, addr, err := noise.ReadMessageFrom(conn)
  49                 if err != nil {
  50                         if err, ok := err.(net.Error); ok && err.Temporary() {
  51                                 continue
  52                         }
  53                         return nil, nil, err
  54                 }
  55                 if msgType == wantedType {
  56                         return msg, addr, nil
  57                 }
  58         }
  59 }
  60
  61 // noiseDial performs a Noise handshake over the given net.PacketConn, and
  62 // returns a noisePacketConn with a working noise.Session.
  63 func noiseDial(conn net.PacketConn, addr net.Addr, pubkey []byte) (*noisePacketConn, error) {
  64         p := []byte{noise.MsgTypeHandshakeInit}
  65         pre, p, err := noise.InitiateHandshake(p, pubkey)
  66         if err != nil {
  67                 return nil, err
  68         }
  69         // TODO: timeout or context
  70         _, err = conn.WriteTo(p, addr)
  71         if err != nil {
  72                 return nil, err
  73         }
  74
  75         msg, _, err := readNoiseMessageOfTypeFrom(conn, noise.MsgTypeHandshakeResp)
  76         if err != nil {
  77                 return nil, err
  78         }
  79
  80         sess, err := pre.FinishHandshake(msg)
  81         if err != nil {
  82                 return nil, err
  83         }
  84
  85         return &noisePacketConn{sess, conn}, nil
  86 }
  87
  88 // ReadFrom implements the net.PacketConn interface for noisePacketConn.
  89 func (c *noisePacketConn) ReadFrom(p []byte) (int, net.Addr, error) {
  90         msg, addr, err := readNoiseMessageOfTypeFrom(c.PacketConn, noise.MsgTypeTransport)
  91         if err != nil {
  92                 return 0, nil, err
  93         }
  94         dec, err := c.sess.Decrypt(nil, msg)
  95         if err != nil {
  96                 return 0, nil, err
  97         }
  98         return copy(p, dec), addr, nil
  99 }
 100
 101 // WriteTo implements the net.PacketConn interface for noisePacketConn.
 102 func (c *noisePacketConn) WriteTo(p []byte, addr net.Addr) (int, error) {
 103         buf := []byte{noise.MsgTypeTransport}
 104         buf, err := c.sess.Encrypt(buf, p)
 105         if err != nil {
 106                 return 0, err
 107         }
 108         return c.PacketConn.WriteTo(buf, addr)
 109 }
 110
 111 func handle(local *net.TCPConn, sess *smux.Session, conv uint32) error {
 112         stream, err := sess.OpenStream()
 113         if err != nil {
 114                 return fmt.Errorf("session %08x opening stream: %v", conv, err)
 115         }
 116         defer func() {
 117                 log.Printf("end stream %08x:%d", conv, stream.ID())
 118                 stream.Close()
 119         }()
 120         log.Printf("begin stream %08x:%d", conv, stream.ID())
 121
 122         var wg sync.WaitGroup
 123         wg.Add(2)
 124         go func() {
 125                 defer wg.Done()
 126                 _, err := io.Copy(stream, local)
 127                 if err == io.EOF {
 128                         // smux Stream.Write may return io.EOF.
 129                         err = nil
 130                 }
 131                 if err != nil && !errors.Is(err, io.ErrClosedPipe) {
 132                         log.Printf("stream %08x:%d copy stream←local: %v", conv, stream.ID(), err)
 133                 }
 134                 local.CloseRead()
 135                 stream.Close()
 136         }()
 137         go func() {
 138                 defer wg.Done()
 139                 _, err := io.Copy(local, stream)
 140                 if err == io.EOF {
 141                         // smux Stream.WriteTo may return io.EOF.
 142                         err = nil
 143                 }
 144                 if err != nil && !errors.Is(err, io.ErrClosedPipe) {
 145                         log.Printf("stream %08x:%d copy local←stream: %v", conv, stream.ID(), err)
 146                 }
 147                 local.CloseWrite()
 148         }()
 149         wg.Wait()
 150
 151         return err
 152 }
 153
 154 func run(serverURL, cacheURL *url.URL, front, localAddr string, pubkey []byte) error {
 155         ln, err := net.Listen("tcp", localAddr)
 156         if err != nil {
 157                 return err
 158         }
 159         defer ln.Close()
 160
 161         http.DefaultTransport.(*http.Transport).MaxConnsPerHost = 20
 162
 163         var poll PollFunc = func(ctx context.Context, p []byte) (io.ReadCloser, error) {
 164                 return exchangeAMP(ctx, serverURL, cacheURL, front, p)
 165         }
 166         pconn := NewPollingPacketConn(turbotunnel.DummyAddr{}, poll)
 167         defer pconn.Close()
 168
 169         // Add a Noise layer over the AMP polling to encrypt and authenticate
 170         // each KCP packet.
 171         nconn, err := noiseDial(pconn, turbotunnel.DummyAddr{}, pubkey)
 172         if err != nil {
 173                 return err
 174         }
 175
 176         // Open a KCP conn over the Noise layer.
 177         conn, err := kcp.NewConn2(turbotunnel.DummyAddr{}, nil, 0, 0, nconn)
 178         if err != nil {
 179                 return fmt.Errorf("opening KCP conn: %v", err)
 180         }
 181         defer func() {
 182                 log.Printf("end session %08x", conn.GetConv())
 183                 conn.Close()
 184         }()
 185         log.Printf("begin session %08x", conn.GetConv())
 186         // Permit coalescing the payloads of consecutive sends.
 187         conn.SetStreamMode(true)
 188         // Disable the dynamic congestion window (limit only by the maximum of
 189         // local and remote static windows).
 190         conn.SetNoDelay(
 191                 0, // default nodelay
 192                 0, // default interval
 193                 0, // default resend
 194                 1, // nc=1 => congestion window off
 195         )
 196         // ACK received data immediately; this is good in our polling model.
 197         conn.SetACKNoDelay(true)
 198         conn.SetWindowSize(1024, 1024) // Default is 32, 32.
 199         // TODO: We could optimize a call to conn.SetMtu here, based on a
 200         // maximum URL length we want to send (such as the 8000 bytes
 201         // recommended at https://datatracker.ietf.org/doc/html/rfc7230#section-3.1.1).
 202         // The idea is that if we can slightly reduce the MTU from its default
 203         // to permit one more packet per request, we should do it.
 204         // E.g. 1400*5 = 7000, but 1320*6 = 7920.
 205
 206         // Start a smux session on the Noise channel.
 207         smuxConfig := smux.DefaultConfig()
 208         smuxConfig.Version = 2
 209         smuxConfig.KeepAliveTimeout = idleTimeout
 210         smuxConfig.MaxReceiveBuffer = 4 * 1024 * 1024 // default is 4 * 1024 * 1024
 211         smuxConfig.MaxStreamBuffer = 1 * 1024 * 1024  // default is 65536
 212         sess, err := smux.Client(conn, smuxConfig)
 213         if err != nil {
 214                 return fmt.Errorf("opening smux session: %v", err)
 215         }
 216         defer sess.Close()
 217
 218         for {
 219                 local, err := ln.Accept()
 220                 if err != nil {
 221                         if err, ok := err.(net.Error); ok && err.Temporary() {
 222                                 continue
 223                         }
 224                         return err
 225                 }
 226                 go func() {
 227                         defer local.Close()
 228                         err := handle(local.(*net.TCPConn), sess, conn.GetConv())
 229                         if err != nil {
 230                                 log.Printf("handle: %v", err)
 231                         }
 232                 }()
 233         }
 234 }
 235
 236 func main() {
 237         var cache string
 238         var front string
 239         var pubkeyFilename string
 240         var pubkeyString string
 241
 242         flag.Usage = func() {
 243                 fmt.Fprintf(flag.CommandLine.Output(), `Usage:
 244   %[1]s -pubkey-file PUBKEYFILE [-cache CACHEURL] [-front DOMAIN] SERVERURL LOCALADDR
 245
 246 Example:
 247   %[1]s -pubkey-file server.pub -cache https://amp.cache.example/ -front amp.cache.example https://server.example/champa/ 127.0.0.1:7000
 248
 249 `, os.Args[0])
 250                 flag.PrintDefaults()
 251         }
 252         flag.StringVar(&cache, "cache", "", "URL of AMP cache (try https://cdn.ampproject.org/)")
 253         flag.StringVar(&front, "front", "", "domain to domain-front HTTPS requests with (try www.google.com)")
 254         flag.StringVar(&pubkeyString, "pubkey", "", fmt.Sprintf("server public key (%d hex digits)", noise.KeyLen*2))
 255         flag.StringVar(&pubkeyFilename, "pubkey-file", "", "read server public key from file")
 256         flag.Parse()
 257
 258         log.SetFlags(log.LstdFlags | log.LUTC)
 259
 260         if flag.NArg() != 2 {
 261                 flag.Usage()
 262                 os.Exit(1)
 263         }
 264         serverURL, err := url.Parse(flag.Arg(0))
 265         if err != nil {
 266                 fmt.Fprintf(os.Stderr, "cannot parse server URL: %v\n", err)
 267                 os.Exit(1)
 268         }
 269         localAddr := flag.Arg(1)
 270
 271         var cacheURL *url.URL
 272         if cache != "" {
 273                 cacheURL, err = url.Parse(cache)
 274                 if err != nil {
 275                         fmt.Fprintf(os.Stderr, "cannot parse AMP cache URL: %v\n", err)
 276                         os.Exit(1)
 277                 }
 278         }
 279
 280         var pubkey []byte
 281         if pubkeyFilename != "" && pubkeyString != "" {
 282                 fmt.Fprintf(os.Stderr, "only one of -pubkey and -pubkey-file may be used\n")
 283                 os.Exit(1)
 284         } else if pubkeyFilename != "" {
 285                 var err error
 286                 pubkey, err = readKeyFromFile(pubkeyFilename)
 287                 if err != nil {
 288                         fmt.Fprintf(os.Stderr, "cannot read pubkey from file: %v\n", err)
 289                         os.Exit(1)
 290                 }
 291         } else if pubkeyString != "" {
 292                 var err error
 293                 pubkey, err = noise.DecodeKey(pubkeyString)
 294                 if err != nil {
 295                         fmt.Fprintf(os.Stderr, "pubkey format error: %v\n", err)
 296                         os.Exit(1)
 297                 }
 298         } else {
 299                 fmt.Fprintf(os.Stderr, "the -pubkey or -pubkey-file option is required\n")
 300                 os.Exit(1)
 301         }
 302
 303         err = run(serverURL, cacheURL, front, localAddr, pubkey)
 304         if err != nil {
 305                 log.Fatal(err)
 306         }
 307 }