client.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475
  1. package mtproto
  2. import (
  3. "context"
  4. "v2ray.com/core"
  5. "v2ray.com/core/common"
  6. "v2ray.com/core/common/buf"
  7. "v2ray.com/core/common/crypto"
  8. "v2ray.com/core/common/net"
  9. "v2ray.com/core/common/task"
  10. "v2ray.com/core/proxy"
  11. )
  12. type Client struct {
  13. }
  14. func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
  15. return &Client{}, nil
  16. }
  17. func (c *Client) Process(ctx context.Context, link *core.Link, dialer proxy.Dialer) error {
  18. dest, ok := proxy.TargetFromContext(ctx)
  19. if !ok {
  20. return newError("unknown destination.")
  21. }
  22. if dest.Network != net.Network_TCP {
  23. return newError("not TCP traffic", dest)
  24. }
  25. conn, err := dialer.Dial(ctx, dest)
  26. if err != nil {
  27. return newError("failed to dial to ", dest).Base(err).AtWarning()
  28. }
  29. defer conn.Close() // nolint: errcheck
  30. auth := NewAuthentication()
  31. defer putAuthenticationObject(auth)
  32. request := func() error {
  33. encryptor := crypto.NewAesCTRStream(auth.EncodingKey[:], auth.EncodingNonce[:])
  34. var header [HeaderSize]byte
  35. encryptor.XORKeyStream(header[:], auth.Header[:])
  36. copy(header[:56], auth.Header[:])
  37. if _, err := conn.Write(header[:]); err != nil {
  38. return newError("failed to write auth header").Base(err)
  39. }
  40. connWriter := buf.NewWriter(crypto.NewCryptionWriter(encryptor, conn))
  41. return buf.Copy(link.Reader, connWriter)
  42. }
  43. response := func() error {
  44. decryptor := crypto.NewAesCTRStream(auth.DecodingKey[:], auth.DecodingNonce[:])
  45. connReader := buf.NewReader(crypto.NewCryptionReader(decryptor, conn))
  46. return buf.Copy(connReader, link.Writer)
  47. }
  48. var responseDoneAndCloseWriter = task.Single(response, task.OnSuccess(task.Close(link.Writer)))
  49. if err := task.Run(task.WithContext(ctx), task.Parallel(request, responseDoneAndCloseWriter))(); err != nil {
  50. return newError("connection ends").Base(err)
  51. }
  52. return nil
  53. }
  54. func init() {
  55. common.Must(common.RegisterConfig((*ClientConfig)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
  56. return NewClient(ctx, config.(*ClientConfig))
  57. }))
  58. }