conn.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467
  1. // Copyright 2011 The Go Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. package ldap
  5. import (
  6. "crypto/tls"
  7. "errors"
  8. "fmt"
  9. "log"
  10. "net"
  11. "sync"
  12. "time"
  13. "gopkg.in/asn1-ber.v1"
  14. )
  15. const (
  16. // MessageQuit causes the processMessages loop to exit
  17. MessageQuit = 0
  18. // MessageRequest sends a request to the server
  19. MessageRequest = 1
  20. // MessageResponse receives a response from the server
  21. MessageResponse = 2
  22. // MessageFinish indicates the client considers a particular message ID to be finished
  23. MessageFinish = 3
  24. // MessageTimeout indicates the client-specified timeout for a particular message ID has been reached
  25. MessageTimeout = 4
  26. )
  27. // PacketResponse contains the packet or error encountered reading a response
  28. type PacketResponse struct {
  29. // Packet is the packet read from the server
  30. Packet *ber.Packet
  31. // Error is an error encountered while reading
  32. Error error
  33. }
  34. // ReadPacket returns the packet or an error
  35. func (pr *PacketResponse) ReadPacket() (*ber.Packet, error) {
  36. if (pr == nil) || (pr.Packet == nil && pr.Error == nil) {
  37. return nil, NewError(ErrorNetwork, errors.New("ldap: could not retrieve response"))
  38. }
  39. return pr.Packet, pr.Error
  40. }
  41. type messageContext struct {
  42. id int64
  43. // close(done) should only be called from finishMessage()
  44. done chan struct{}
  45. // close(responses) should only be called from processMessages(), and only sent to from sendResponse()
  46. responses chan *PacketResponse
  47. }
  48. // sendResponse should only be called within the processMessages() loop which
  49. // is also responsible for closing the responses channel.
  50. func (msgCtx *messageContext) sendResponse(packet *PacketResponse) {
  51. select {
  52. case msgCtx.responses <- packet:
  53. // Successfully sent packet to message handler.
  54. case <-msgCtx.done:
  55. // The request handler is done and will not receive more
  56. // packets.
  57. }
  58. }
  59. type messagePacket struct {
  60. Op int
  61. MessageID int64
  62. Packet *ber.Packet
  63. Context *messageContext
  64. }
  65. type sendMessageFlags uint
  66. const (
  67. startTLS sendMessageFlags = 1 << iota
  68. )
  69. // Conn represents an LDAP Connection
  70. type Conn struct {
  71. conn net.Conn
  72. isTLS bool
  73. isClosing bool
  74. closeErr error
  75. isStartingTLS bool
  76. Debug debugging
  77. chanConfirm chan bool
  78. messageContexts map[int64]*messageContext
  79. chanMessage chan *messagePacket
  80. chanMessageID chan int64
  81. wgSender sync.WaitGroup
  82. wgClose sync.WaitGroup
  83. once sync.Once
  84. outstandingRequests uint
  85. messageMutex sync.Mutex
  86. requestTimeout time.Duration
  87. }
  88. var _ Client = &Conn{}
  89. // DefaultTimeout is a package-level variable that sets the timeout value
  90. // used for the Dial and DialTLS methods.
  91. //
  92. // WARNING: since this is a package-level variable, setting this value from
  93. // multiple places will probably result in undesired behaviour.
  94. var DefaultTimeout = 60 * time.Second
  95. // Dial connects to the given address on the given network using net.Dial
  96. // and then returns a new Conn for the connection.
  97. func Dial(network, addr string) (*Conn, error) {
  98. c, err := net.DialTimeout(network, addr, DefaultTimeout)
  99. if err != nil {
  100. return nil, NewError(ErrorNetwork, err)
  101. }
  102. conn := NewConn(c, false)
  103. conn.Start()
  104. return conn, nil
  105. }
  106. // DialTLS connects to the given address on the given network using tls.Dial
  107. // and then returns a new Conn for the connection.
  108. func DialTLS(network, addr string, config *tls.Config) (*Conn, error) {
  109. dc, err := net.DialTimeout(network, addr, DefaultTimeout)
  110. if err != nil {
  111. return nil, NewError(ErrorNetwork, err)
  112. }
  113. c := tls.Client(dc, config)
  114. err = c.Handshake()
  115. if err != nil {
  116. // Handshake error, close the established connection before we return an error
  117. dc.Close()
  118. return nil, NewError(ErrorNetwork, err)
  119. }
  120. conn := NewConn(c, true)
  121. conn.Start()
  122. return conn, nil
  123. }
  124. // NewConn returns a new Conn using conn for network I/O.
  125. func NewConn(conn net.Conn, isTLS bool) *Conn {
  126. return &Conn{
  127. conn: conn,
  128. chanConfirm: make(chan bool),
  129. chanMessageID: make(chan int64),
  130. chanMessage: make(chan *messagePacket, 10),
  131. messageContexts: map[int64]*messageContext{},
  132. requestTimeout: 0,
  133. isTLS: isTLS,
  134. }
  135. }
  136. // Start initializes goroutines to read responses and process messages
  137. func (l *Conn) Start() {
  138. go l.reader()
  139. go l.processMessages()
  140. l.wgClose.Add(1)
  141. }
  142. // Close closes the connection.
  143. func (l *Conn) Close() {
  144. l.once.Do(func() {
  145. l.isClosing = true
  146. l.wgSender.Wait()
  147. l.Debug.Printf("Sending quit message and waiting for confirmation")
  148. l.chanMessage <- &messagePacket{Op: MessageQuit}
  149. <-l.chanConfirm
  150. close(l.chanMessage)
  151. l.Debug.Printf("Closing network connection")
  152. if err := l.conn.Close(); err != nil {
  153. log.Print(err)
  154. }
  155. l.wgClose.Done()
  156. })
  157. l.wgClose.Wait()
  158. }
  159. // SetTimeout sets the time after a request is sent that a MessageTimeout triggers
  160. func (l *Conn) SetTimeout(timeout time.Duration) {
  161. if timeout > 0 {
  162. l.requestTimeout = timeout
  163. }
  164. }
  165. // Returns the next available messageID
  166. func (l *Conn) nextMessageID() int64 {
  167. if l.chanMessageID != nil {
  168. if messageID, ok := <-l.chanMessageID; ok {
  169. return messageID
  170. }
  171. }
  172. return 0
  173. }
  174. // StartTLS sends the command to start a TLS session and then creates a new TLS Client
  175. func (l *Conn) StartTLS(config *tls.Config) error {
  176. if l.isTLS {
  177. return NewError(ErrorNetwork, errors.New("ldap: already encrypted"))
  178. }
  179. packet := ber.Encode(ber.ClassUniversal, ber.TypeConstructed, ber.TagSequence, nil, "LDAP Request")
  180. packet.AppendChild(ber.NewInteger(ber.ClassUniversal, ber.TypePrimitive, ber.TagInteger, l.nextMessageID(), "MessageID"))
  181. request := ber.Encode(ber.ClassApplication, ber.TypeConstructed, ApplicationExtendedRequest, nil, "Start TLS")
  182. request.AppendChild(ber.NewString(ber.ClassContext, ber.TypePrimitive, 0, "1.3.6.1.4.1.1466.20037", "TLS Extended Command"))
  183. packet.AppendChild(request)
  184. l.Debug.PrintPacket(packet)
  185. msgCtx, err := l.sendMessageWithFlags(packet, startTLS)
  186. if err != nil {
  187. return err
  188. }
  189. defer l.finishMessage(msgCtx)
  190. l.Debug.Printf("%d: waiting for response", msgCtx.id)
  191. packetResponse, ok := <-msgCtx.responses
  192. if !ok {
  193. return NewError(ErrorNetwork, errors.New("ldap: response channel closed"))
  194. }
  195. packet, err = packetResponse.ReadPacket()
  196. l.Debug.Printf("%d: got response %p", msgCtx.id, packet)
  197. if err != nil {
  198. return err
  199. }
  200. if l.Debug {
  201. if err := addLDAPDescriptions(packet); err != nil {
  202. l.Close()
  203. return err
  204. }
  205. ber.PrintPacket(packet)
  206. }
  207. if resultCode, message := getLDAPResultCode(packet); resultCode == LDAPResultSuccess {
  208. conn := tls.Client(l.conn, config)
  209. if err := conn.Handshake(); err != nil {
  210. l.Close()
  211. return NewError(ErrorNetwork, fmt.Errorf("TLS handshake failed (%v)", err))
  212. }
  213. l.isTLS = true
  214. l.conn = conn
  215. } else {
  216. return NewError(resultCode, fmt.Errorf("ldap: cannot StartTLS (%s)", message))
  217. }
  218. go l.reader()
  219. return nil
  220. }
  221. func (l *Conn) sendMessage(packet *ber.Packet) (*messageContext, error) {
  222. return l.sendMessageWithFlags(packet, 0)
  223. }
  224. func (l *Conn) sendMessageWithFlags(packet *ber.Packet, flags sendMessageFlags) (*messageContext, error) {
  225. if l.isClosing {
  226. return nil, NewError(ErrorNetwork, errors.New("ldap: connection closed"))
  227. }
  228. l.messageMutex.Lock()
  229. l.Debug.Printf("flags&startTLS = %d", flags&startTLS)
  230. if l.isStartingTLS {
  231. l.messageMutex.Unlock()
  232. return nil, NewError(ErrorNetwork, errors.New("ldap: connection is in startls phase"))
  233. }
  234. if flags&startTLS != 0 {
  235. if l.outstandingRequests != 0 {
  236. l.messageMutex.Unlock()
  237. return nil, NewError(ErrorNetwork, errors.New("ldap: cannot StartTLS with outstanding requests"))
  238. }
  239. l.isStartingTLS = true
  240. }
  241. l.outstandingRequests++
  242. l.messageMutex.Unlock()
  243. responses := make(chan *PacketResponse)
  244. messageID := packet.Children[0].Value.(int64)
  245. message := &messagePacket{
  246. Op: MessageRequest,
  247. MessageID: messageID,
  248. Packet: packet,
  249. Context: &messageContext{
  250. id: messageID,
  251. done: make(chan struct{}),
  252. responses: responses,
  253. },
  254. }
  255. l.sendProcessMessage(message)
  256. return message.Context, nil
  257. }
  258. func (l *Conn) finishMessage(msgCtx *messageContext) {
  259. close(msgCtx.done)
  260. if l.isClosing {
  261. return
  262. }
  263. l.messageMutex.Lock()
  264. l.outstandingRequests--
  265. if l.isStartingTLS {
  266. l.isStartingTLS = false
  267. }
  268. l.messageMutex.Unlock()
  269. message := &messagePacket{
  270. Op: MessageFinish,
  271. MessageID: msgCtx.id,
  272. }
  273. l.sendProcessMessage(message)
  274. }
  275. func (l *Conn) sendProcessMessage(message *messagePacket) bool {
  276. if l.isClosing {
  277. return false
  278. }
  279. l.wgSender.Add(1)
  280. l.chanMessage <- message
  281. l.wgSender.Done()
  282. return true
  283. }
  284. func (l *Conn) processMessages() {
  285. defer func() {
  286. if err := recover(); err != nil {
  287. log.Printf("ldap: recovered panic in processMessages: %v", err)
  288. }
  289. for messageID, msgCtx := range l.messageContexts {
  290. // If we are closing due to an error, inform anyone who
  291. // is waiting about the error.
  292. if l.isClosing && l.closeErr != nil {
  293. msgCtx.sendResponse(&PacketResponse{Error: l.closeErr})
  294. }
  295. l.Debug.Printf("Closing channel for MessageID %d", messageID)
  296. close(msgCtx.responses)
  297. delete(l.messageContexts, messageID)
  298. }
  299. close(l.chanMessageID)
  300. l.chanConfirm <- true
  301. close(l.chanConfirm)
  302. }()
  303. var messageID int64 = 1
  304. for {
  305. select {
  306. case l.chanMessageID <- messageID:
  307. messageID++
  308. case message, ok := <-l.chanMessage:
  309. if !ok {
  310. l.Debug.Printf("Shutting down - message channel is closed")
  311. return
  312. }
  313. switch message.Op {
  314. case MessageQuit:
  315. l.Debug.Printf("Shutting down - quit message received")
  316. return
  317. case MessageRequest:
  318. // Add to message list and write to network
  319. l.Debug.Printf("Sending message %d", message.MessageID)
  320. buf := message.Packet.Bytes()
  321. _, err := l.conn.Write(buf)
  322. if err != nil {
  323. l.Debug.Printf("Error Sending Message: %s", err.Error())
  324. message.Context.sendResponse(&PacketResponse{Error: fmt.Errorf("unable to send request: %s", err)})
  325. close(message.Context.responses)
  326. break
  327. }
  328. // Only add to messageContexts if we were able to
  329. // successfully write the message.
  330. l.messageContexts[message.MessageID] = message.Context
  331. // Add timeout if defined
  332. if l.requestTimeout > 0 {
  333. go func() {
  334. defer func() {
  335. if err := recover(); err != nil {
  336. log.Printf("ldap: recovered panic in RequestTimeout: %v", err)
  337. }
  338. }()
  339. time.Sleep(l.requestTimeout)
  340. timeoutMessage := &messagePacket{
  341. Op: MessageTimeout,
  342. MessageID: message.MessageID,
  343. }
  344. l.sendProcessMessage(timeoutMessage)
  345. }()
  346. }
  347. case MessageResponse:
  348. l.Debug.Printf("Receiving message %d", message.MessageID)
  349. if msgCtx, ok := l.messageContexts[message.MessageID]; ok {
  350. msgCtx.sendResponse(&PacketResponse{message.Packet, nil})
  351. } else {
  352. log.Printf("Received unexpected message %d, %v", message.MessageID, l.isClosing)
  353. ber.PrintPacket(message.Packet)
  354. }
  355. case MessageTimeout:
  356. // Handle the timeout by closing the channel
  357. // All reads will return immediately
  358. if msgCtx, ok := l.messageContexts[message.MessageID]; ok {
  359. l.Debug.Printf("Receiving message timeout for %d", message.MessageID)
  360. msgCtx.sendResponse(&PacketResponse{message.Packet, errors.New("ldap: connection timed out")})
  361. delete(l.messageContexts, message.MessageID)
  362. close(msgCtx.responses)
  363. }
  364. case MessageFinish:
  365. l.Debug.Printf("Finished message %d", message.MessageID)
  366. if msgCtx, ok := l.messageContexts[message.MessageID]; ok {
  367. delete(l.messageContexts, message.MessageID)
  368. close(msgCtx.responses)
  369. }
  370. }
  371. }
  372. }
  373. }
  374. func (l *Conn) reader() {
  375. cleanstop := false
  376. defer func() {
  377. if err := recover(); err != nil {
  378. log.Printf("ldap: recovered panic in reader: %v", err)
  379. }
  380. if !cleanstop {
  381. l.Close()
  382. }
  383. }()
  384. for {
  385. if cleanstop {
  386. l.Debug.Printf("reader clean stopping (without closing the connection)")
  387. return
  388. }
  389. packet, err := ber.ReadPacket(l.conn)
  390. if err != nil {
  391. // A read error is expected here if we are closing the connection...
  392. if !l.isClosing {
  393. l.closeErr = fmt.Errorf("unable to read LDAP response packet: %s", err)
  394. l.Debug.Printf("reader error: %s", err.Error())
  395. }
  396. return
  397. }
  398. addLDAPDescriptions(packet)
  399. if len(packet.Children) == 0 {
  400. l.Debug.Printf("Received bad ldap packet")
  401. continue
  402. }
  403. l.messageMutex.Lock()
  404. if l.isStartingTLS {
  405. cleanstop = true
  406. }
  407. l.messageMutex.Unlock()
  408. message := &messagePacket{
  409. Op: MessageResponse,
  410. MessageID: packet.Children[0].Value.(int64),
  411. Packet: packet,
  412. }
  413. if !l.sendProcessMessage(message) {
  414. return
  415. }
  416. }
  417. }