anonymous group photoblog software
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

1338 lines
43 KiB

  1. # Copyright (c) 2002, 2003, 2005, 2006 Allan Saddi <allan@saddi.com>
  2. # All rights reserved.
  3. #
  4. # Redistribution and use in source and binary forms, with or without
  5. # modification, are permitted provided that the following conditions
  6. # are met:
  7. # 1. Redistributions of source code must retain the above copyright
  8. # notice, this list of conditions and the following disclaimer.
  9. # 2. Redistributions in binary form must reproduce the above copyright
  10. # notice, this list of conditions and the following disclaimer in the
  11. # documentation and/or other materials provided with the distribution.
  12. #
  13. # THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
  14. # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  15. # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  16. # ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
  17. # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  18. # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  19. # OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  20. # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  21. # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  22. # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  23. # SUCH DAMAGE.
  24. #
  25. # $Id$
  26. """
  27. fcgi - a FastCGI/WSGI gateway.
  28. For more information about FastCGI, see <http://www.fastcgi.com/>.
  29. For more information about the Web Server Gateway Interface, see
  30. <http://www.python.org/peps/pep-0333.html>.
  31. Example usage:
  32. #!/usr/bin/env python
  33. from myapplication import app # Assume app is your WSGI application object
  34. from fcgi import WSGIServer
  35. WSGIServer(app).run()
  36. See the documentation for WSGIServer/Server for more information.
  37. On most platforms, fcgi will fallback to regular CGI behavior if run in a
  38. non-FastCGI context. If you want to force CGI behavior, set the environment
  39. variable FCGI_FORCE_CGI to "Y" or "y".
  40. """
  41. import config, config_defaults
  42. __author__ = 'Allan Saddi <allan@saddi.com>'
  43. __version__ = '$Revision$'
  44. import sys
  45. import os
  46. import signal
  47. import struct
  48. import cStringIO as StringIO
  49. import select
  50. import socket
  51. import errno
  52. import traceback
  53. try:
  54. import thread
  55. import threading
  56. thread_available = True
  57. except ImportError:
  58. import dummy_thread as thread
  59. import dummy_threading as threading
  60. thread_available = False
  61. # Apparently 2.3 doesn't define SHUT_WR? Assume it is 1 in this case.
  62. if not hasattr(socket, 'SHUT_WR'):
  63. socket.SHUT_WR = 1
  64. __all__ = ['WSGIServer']
  65. # Constants from the spec.
  66. FCGI_LISTENSOCK_FILENO = 0
  67. FCGI_HEADER_LEN = 8
  68. FCGI_VERSION_1 = 1
  69. FCGI_BEGIN_REQUEST = 1
  70. FCGI_ABORT_REQUEST = 2
  71. FCGI_END_REQUEST = 3
  72. FCGI_PARAMS = 4
  73. FCGI_STDIN = 5
  74. FCGI_STDOUT = 6
  75. FCGI_STDERR = 7
  76. FCGI_DATA = 8
  77. FCGI_GET_VALUES = 9
  78. FCGI_GET_VALUES_RESULT = 10
  79. FCGI_UNKNOWN_TYPE = 11
  80. FCGI_MAXTYPE = FCGI_UNKNOWN_TYPE
  81. FCGI_NULL_REQUEST_ID = 0
  82. FCGI_KEEP_CONN = 1
  83. FCGI_RESPONDER = 1
  84. FCGI_AUTHORIZER = 2
  85. FCGI_FILTER = 3
  86. FCGI_REQUEST_COMPLETE = 0
  87. FCGI_CANT_MPX_CONN = 1
  88. FCGI_OVERLOADED = 2
  89. FCGI_UNKNOWN_ROLE = 3
  90. FCGI_MAX_CONNS = 'FCGI_MAX_CONNS'
  91. FCGI_MAX_REQS = 'FCGI_MAX_REQS'
  92. FCGI_MPXS_CONNS = 'FCGI_MPXS_CONNS'
  93. FCGI_Header = '!BBHHBx'
  94. FCGI_BeginRequestBody = '!HB5x'
  95. FCGI_EndRequestBody = '!LB3x'
  96. FCGI_UnknownTypeBody = '!B7x'
  97. FCGI_EndRequestBody_LEN = struct.calcsize(FCGI_EndRequestBody)
  98. FCGI_UnknownTypeBody_LEN = struct.calcsize(FCGI_UnknownTypeBody)
  99. if __debug__:
  100. import time
  101. # Set non-zero to write debug output to a file.
  102. DEBUG = 0
  103. DEBUGLOG = '/tmp/fcgi.log'
  104. def _debug(level, msg):
  105. if DEBUG < level:
  106. return
  107. try:
  108. f = open(DEBUGLOG, 'a')
  109. f.write('%sfcgi: %s\n' % (time.ctime()[4:-4], msg))
  110. f.close()
  111. except:
  112. pass
  113. class InputStream(object):
  114. """
  115. File-like object representing FastCGI input streams (FCGI_STDIN and
  116. FCGI_DATA). Supports the minimum methods required by WSGI spec.
  117. """
  118. def __init__(self, conn):
  119. self._conn = conn
  120. # See Server.
  121. self._shrinkThreshold = conn.server.inputStreamShrinkThreshold
  122. self._buf = ''
  123. self._bufList = []
  124. self._pos = 0 # Current read position.
  125. self._avail = 0 # Number of bytes currently available.
  126. self._eof = False # True when server has sent EOF notification.
  127. def _shrinkBuffer(self):
  128. """Gets rid of already read data (since we can't rewind)."""
  129. if self._pos >= self._shrinkThreshold:
  130. self._buf = self._buf[self._pos:]
  131. self._avail -= self._pos
  132. self._pos = 0
  133. assert self._avail >= 0
  134. def _waitForData(self):
  135. """Waits for more data to become available."""
  136. self._conn.process_input()
  137. def read(self, n=-1):
  138. if self._pos == self._avail and self._eof:
  139. return ''
  140. while True:
  141. if n < 0 or (self._avail - self._pos) < n:
  142. # Not enough data available.
  143. if self._eof:
  144. # And there's no more coming.
  145. newPos = self._avail
  146. break
  147. else:
  148. # Wait for more data.
  149. self._waitForData()
  150. continue
  151. else:
  152. newPos = self._pos + n
  153. break
  154. # Merge buffer list, if necessary.
  155. if self._bufList:
  156. self._buf += ''.join(self._bufList)
  157. self._bufList = []
  158. r = self._buf[self._pos:newPos]
  159. self._pos = newPos
  160. self._shrinkBuffer()
  161. return r
  162. def readline(self, length=None):
  163. if self._pos == self._avail and self._eof:
  164. return ''
  165. while True:
  166. # Unfortunately, we need to merge the buffer list early.
  167. if self._bufList:
  168. self._buf += ''.join(self._bufList)
  169. self._bufList = []
  170. # Find newline.
  171. i = self._buf.find('\n', self._pos)
  172. if i < 0:
  173. # Not found?
  174. if self._eof:
  175. # No more data coming.
  176. newPos = self._avail
  177. break
  178. else:
  179. # Wait for more to come.
  180. self._waitForData()
  181. continue
  182. else:
  183. newPos = i + 1
  184. break
  185. if length is not None:
  186. if self._pos + length < newPos:
  187. newPos = self._pos + length
  188. r = self._buf[self._pos:newPos]
  189. self._pos = newPos
  190. self._shrinkBuffer()
  191. return r
  192. def readlines(self, sizehint=0):
  193. total = 0
  194. lines = []
  195. line = self.readline()
  196. while line:
  197. lines.append(line)
  198. total += len(line)
  199. if 0 < sizehint <= total:
  200. break
  201. line = self.readline()
  202. return lines
  203. def __iter__(self):
  204. return self
  205. def next(self):
  206. r = self.readline()
  207. if not r:
  208. raise StopIteration
  209. return r
  210. def add_data(self, data):
  211. if not data:
  212. self._eof = True
  213. else:
  214. self._bufList.append(data)
  215. self._avail += len(data)
  216. class MultiplexedInputStream(InputStream):
  217. """
  218. A version of InputStream meant to be used with MultiplexedConnections.
  219. Assumes the MultiplexedConnection (the producer) and the Request
  220. (the consumer) are running in different threads.
  221. """
  222. def __init__(self, conn):
  223. super(MultiplexedInputStream, self).__init__(conn)
  224. # Arbitrates access to this InputStream (it's used simultaneously
  225. # by a Request and its owning Connection object).
  226. lock = threading.RLock()
  227. # Notifies Request thread that there is new data available.
  228. self._lock = threading.Condition(lock)
  229. def _waitForData(self):
  230. # Wait for notification from add_data().
  231. self._lock.wait()
  232. def read(self, n=-1):
  233. self._lock.acquire()
  234. try:
  235. return super(MultiplexedInputStream, self).read(n)
  236. finally:
  237. self._lock.release()
  238. def readline(self, length=None):
  239. self._lock.acquire()
  240. try:
  241. return super(MultiplexedInputStream, self).readline(length)
  242. finally:
  243. self._lock.release()
  244. def add_data(self, data):
  245. self._lock.acquire()
  246. try:
  247. super(MultiplexedInputStream, self).add_data(data)
  248. self._lock.notify()
  249. finally:
  250. self._lock.release()
  251. class OutputStream(object):
  252. """
  253. FastCGI output stream (FCGI_STDOUT/FCGI_STDERR). By default, calls to
  254. write() or writelines() immediately result in Records being sent back
  255. to the server. Buffering should be done in a higher level!
  256. """
  257. def __init__(self, conn, req, type, buffered=False):
  258. self._conn = conn
  259. self._req = req
  260. self._type = type
  261. self._buffered = buffered
  262. self._bufList = [] # Used if buffered is True
  263. self.dataWritten = False
  264. self.closed = False
  265. def _write(self, data):
  266. length = len(data)
  267. while length:
  268. toWrite = min(length, self._req.server.maxwrite - FCGI_HEADER_LEN)
  269. rec = Record(self._type, self._req.requestId)
  270. rec.contentLength = toWrite
  271. rec.contentData = data[:toWrite]
  272. self._conn.writeRecord(rec)
  273. data = data[toWrite:]
  274. length -= toWrite
  275. def write(self, data):
  276. assert not self.closed
  277. if not data:
  278. return
  279. self.dataWritten = True
  280. if self._buffered:
  281. self._bufList.append(data)
  282. else:
  283. self._write(data)
  284. def writelines(self, lines):
  285. assert not self.closed
  286. for line in lines:
  287. self.write(line)
  288. def flush(self):
  289. # Only need to flush if this OutputStream is actually buffered.
  290. if self._buffered:
  291. data = ''.join(self._bufList)
  292. self._bufList = []
  293. self._write(data)
  294. # Though available, the following should NOT be called by WSGI apps.
  295. def close(self):
  296. """Sends end-of-stream notification, if necessary."""
  297. if not self.closed and self.dataWritten:
  298. self.flush()
  299. rec = Record(self._type, self._req.requestId)
  300. self._conn.writeRecord(rec)
  301. self.closed = True
  302. class TeeOutputStream(object):
  303. """
  304. Simple wrapper around two or more output file-like objects that copies
  305. written data to all streams.
  306. """
  307. def __init__(self, streamList):
  308. self._streamList = streamList
  309. def write(self, data):
  310. for f in self._streamList:
  311. f.write(data)
  312. def writelines(self, lines):
  313. for line in lines:
  314. self.write(line)
  315. def flush(self):
  316. for f in self._streamList:
  317. f.flush()
  318. class StdoutWrapper(object):
  319. """
  320. Wrapper for sys.stdout so we know if data has actually been written.
  321. """
  322. def __init__(self, stdout):
  323. self._file = stdout
  324. self.dataWritten = False
  325. def write(self, data):
  326. if data:
  327. self.dataWritten = True
  328. self._file.write(data)
  329. def writelines(self, lines):
  330. for line in lines:
  331. self.write(line)
  332. def __getattr__(self, name):
  333. return getattr(self._file, name)
  334. def decode_pair(s, pos=0):
  335. """
  336. Decodes a name/value pair.
  337. The number of bytes decoded as well as the name/value pair
  338. are returned.
  339. """
  340. nameLength = ord(s[pos])
  341. if nameLength & 128:
  342. nameLength = struct.unpack('!L', s[pos:pos+4])[0] & 0x7fffffff
  343. pos += 4
  344. else:
  345. pos += 1
  346. valueLength = ord(s[pos])
  347. if valueLength & 128:
  348. valueLength = struct.unpack('!L', s[pos:pos+4])[0] & 0x7fffffff
  349. pos += 4
  350. else:
  351. pos += 1
  352. name = s[pos:pos+nameLength]
  353. pos += nameLength
  354. value = s[pos:pos+valueLength]
  355. pos += valueLength
  356. return (pos, (name, value))
  357. def encode_pair(name, value):
  358. """
  359. Encodes a name/value pair.
  360. The encoded string is returned.
  361. """
  362. nameLength = len(name)
  363. if nameLength < 128:
  364. s = chr(nameLength)
  365. else:
  366. s = struct.pack('!L', nameLength | 0x80000000L)
  367. valueLength = len(value)
  368. if valueLength < 128:
  369. s += chr(valueLength)
  370. else:
  371. s += struct.pack('!L', valueLength | 0x80000000L)
  372. return s + name + value
  373. class Record(object):
  374. """
  375. A FastCGI Record.
  376. Used for encoding/decoding records.
  377. """
  378. def __init__(self, type=FCGI_UNKNOWN_TYPE, requestId=FCGI_NULL_REQUEST_ID):
  379. self.version = FCGI_VERSION_1
  380. self.type = type
  381. self.requestId = requestId
  382. self.contentLength = 0
  383. self.paddingLength = 0
  384. self.contentData = ''
  385. def _recvall(sock, length):
  386. """
  387. Attempts to receive length bytes from a socket, blocking if necessary.
  388. (Socket may be blocking or non-blocking.)
  389. """
  390. dataList = []
  391. recvLen = 0
  392. while length:
  393. try:
  394. data = sock.recv(length)
  395. except socket.error, e:
  396. if e[0] == errno.EAGAIN:
  397. select.select([sock], [], [])
  398. continue
  399. else:
  400. raise
  401. if not data: # EOF
  402. break
  403. dataList.append(data)
  404. dataLen = len(data)
  405. recvLen += dataLen
  406. length -= dataLen
  407. return ''.join(dataList), recvLen
  408. _recvall = staticmethod(_recvall)
  409. def read(self, sock):
  410. """Read and decode a Record from a socket."""
  411. try:
  412. header, length = self._recvall(sock, FCGI_HEADER_LEN)
  413. except:
  414. raise EOFError
  415. if length < FCGI_HEADER_LEN:
  416. raise EOFError
  417. self.version, self.type, self.requestId, self.contentLength, \
  418. self.paddingLength = struct.unpack(FCGI_Header, header)
  419. if __debug__: _debug(9, 'read: fd = %d, type = %d, requestId = %d, '
  420. 'contentLength = %d' %
  421. (sock.fileno(), self.type, self.requestId,
  422. self.contentLength))
  423. if self.contentLength:
  424. try:
  425. self.contentData, length = self._recvall(sock,
  426. self.contentLength)
  427. except:
  428. raise EOFError
  429. if length < self.contentLength:
  430. raise EOFError
  431. if self.paddingLength:
  432. try:
  433. self._recvall(sock, self.paddingLength)
  434. except:
  435. raise EOFError
  436. def _sendall(sock, data):
  437. """
  438. Writes data to a socket and does not return until all the data is sent.
  439. """
  440. length = len(data)
  441. while length:
  442. try:
  443. sent = sock.send(data)
  444. except socket.error, e:
  445. if e[0] == errno.EAGAIN:
  446. select.select([], [sock], [])
  447. continue
  448. else:
  449. raise
  450. data = data[sent:]
  451. length -= sent
  452. _sendall = staticmethod(_sendall)
  453. def write(self, sock):
  454. """Encode and write a Record to a socket."""
  455. self.paddingLength = -self.contentLength & 7
  456. if __debug__: _debug(9, 'write: fd = %d, type = %d, requestId = %d, '
  457. 'contentLength = %d' %
  458. (sock.fileno(), self.type, self.requestId,
  459. self.contentLength))
  460. header = struct.pack(FCGI_Header, self.version, self.type,
  461. self.requestId, self.contentLength,
  462. self.paddingLength)
  463. self._sendall(sock, header)
  464. if self.contentLength:
  465. self._sendall(sock, self.contentData)
  466. if self.paddingLength:
  467. self._sendall(sock, '\x00'*self.paddingLength)
  468. class Request(object):
  469. """
  470. Represents a single FastCGI request.
  471. These objects are passed to your handler and is the main interface
  472. between your handler and the fcgi module. The methods should not
  473. be called by your handler. However, server, params, stdin, stdout,
  474. stderr, and data are free for your handler's use.
  475. """
  476. def __init__(self, conn, inputStreamClass):
  477. self._conn = conn
  478. self.server = conn.server
  479. self.params = {}
  480. self.stdin = inputStreamClass(conn)
  481. self.stdout = OutputStream(conn, self, FCGI_STDOUT)
  482. self.stderr = OutputStream(conn, self, FCGI_STDERR, buffered=True)
  483. self.data = inputStreamClass(conn)
  484. def run(self):
  485. """Runs the handler, flushes the streams, and ends the request."""
  486. try:
  487. protocolStatus, appStatus = self.server.handler(self)
  488. except:
  489. traceback.print_exc(file=self.stderr)
  490. self.stderr.flush()
  491. if not self.stdout.dataWritten:
  492. self.server.error(self)
  493. protocolStatus, appStatus = FCGI_REQUEST_COMPLETE, 0
  494. if __debug__: _debug(1, 'protocolStatus = %d, appStatus = %d' %
  495. (protocolStatus, appStatus))
  496. self._flush()
  497. self._end(appStatus, protocolStatus)
  498. def _end(self, appStatus=0L, protocolStatus=FCGI_REQUEST_COMPLETE):
  499. self._conn.end_request(self, appStatus, protocolStatus)
  500. def _flush(self):
  501. self.stdout.close()
  502. self.stderr.close()
  503. class CGIRequest(Request):
  504. """A normal CGI request disguised as a FastCGI request."""
  505. def __init__(self, server):
  506. # These are normally filled in by Connection.
  507. self.requestId = 1
  508. self.role = FCGI_RESPONDER
  509. self.flags = 0
  510. self.aborted = False
  511. self.server = server
  512. self.params = dict(os.environ)
  513. self.stdin = sys.stdin
  514. self.stdout = StdoutWrapper(sys.stdout) # Oh, the humanity!
  515. self.stderr = sys.stderr
  516. self.data = StringIO.StringIO()
  517. def _end(self, appStatus=0L, protocolStatus=FCGI_REQUEST_COMPLETE):
  518. sys.exit(appStatus)
  519. def _flush(self):
  520. # Not buffered, do nothing.
  521. pass
  522. class Connection(object):
  523. """
  524. A Connection with the web server.
  525. Each Connection is associated with a single socket (which is
  526. connected to the web server) and is responsible for handling all
  527. the FastCGI message processing for that socket.
  528. """
  529. _multiplexed = False
  530. _inputStreamClass = InputStream
  531. def __init__(self, sock, addr, server):
  532. self._sock = sock
  533. self._addr = addr
  534. self.server = server
  535. # Active Requests for this Connection, mapped by request ID.
  536. self._requests = {}
  537. def _cleanupSocket(self):
  538. """Close the Connection's socket."""
  539. try:
  540. self._sock.shutdown(socket.SHUT_WR)
  541. except:
  542. return
  543. try:
  544. while True:
  545. r, w, e = select.select([self._sock], [], [])
  546. if not r or not self._sock.recv(1024):
  547. break
  548. except:
  549. pass
  550. self._sock.close()
  551. def run(self):
  552. """Begin processing data from the socket."""
  553. self._keepGoing = True
  554. while self._keepGoing:
  555. try:
  556. self.process_input()
  557. except EOFError:
  558. break
  559. except (select.error, socket.error), e:
  560. if e[0] == errno.EBADF: # Socket was closed by Request.
  561. break
  562. raise
  563. self._cleanupSocket()
  564. def process_input(self):
  565. """Attempt to read a single Record from the socket and process it."""
  566. # Currently, any children Request threads notify this Connection
  567. # that it is no longer needed by closing the Connection's socket.
  568. # We need to put a timeout on select, otherwise we might get
  569. # stuck in it indefinitely... (I don't like this solution.)
  570. while self._keepGoing:
  571. try:
  572. r, w, e = select.select([self._sock], [], [], 1.0)
  573. except ValueError:
  574. # Sigh. ValueError gets thrown sometimes when passing select
  575. # a closed socket.
  576. raise EOFError
  577. if r: break
  578. if not self._keepGoing:
  579. return
  580. rec = Record()
  581. rec.read(self._sock)
  582. if rec.type == FCGI_GET_VALUES:
  583. self._do_get_values(rec)
  584. elif rec.type == FCGI_BEGIN_REQUEST:
  585. self._do_begin_request(rec)
  586. elif rec.type == FCGI_ABORT_REQUEST:
  587. self._do_abort_request(rec)
  588. elif rec.type == FCGI_PARAMS:
  589. self._do_params(rec)
  590. elif rec.type == FCGI_STDIN:
  591. self._do_stdin(rec)
  592. elif rec.type == FCGI_DATA:
  593. self._do_data(rec)
  594. elif rec.requestId == FCGI_NULL_REQUEST_ID:
  595. self._do_unknown_type(rec)
  596. else:
  597. # Need to complain about this.
  598. pass
  599. def writeRecord(self, rec):
  600. """
  601. Write a Record to the socket.
  602. """
  603. rec.write(self._sock)
  604. def end_request(self, req, appStatus=0L,
  605. protocolStatus=FCGI_REQUEST_COMPLETE, remove=True):
  606. """
  607. End a Request.
  608. Called by Request objects. An FCGI_END_REQUEST Record is
  609. sent to the web server. If the web server no longer requires
  610. the connection, the socket is closed, thereby ending this
  611. Connection (run() returns).
  612. """
  613. rec = Record(FCGI_END_REQUEST, req.requestId)
  614. rec.contentData = struct.pack(FCGI_EndRequestBody, appStatus,
  615. protocolStatus)
  616. rec.contentLength = FCGI_EndRequestBody_LEN
  617. self.writeRecord(rec)
  618. if remove:
  619. del self._requests[req.requestId]
  620. if __debug__: _debug(2, 'end_request: flags = %d' % req.flags)
  621. if not (req.flags & FCGI_KEEP_CONN) and not self._requests:
  622. self._cleanupSocket()
  623. self._keepGoing = False
  624. def _do_get_values(self, inrec):
  625. """Handle an FCGI_GET_VALUES request from the web server."""
  626. outrec = Record(FCGI_GET_VALUES_RESULT)
  627. pos = 0
  628. while pos < inrec.contentLength:
  629. pos, (name, value) = decode_pair(inrec.contentData, pos)
  630. cap = self.server.capability.get(name)
  631. if cap is not None:
  632. outrec.contentData += encode_pair(name, str(cap))
  633. outrec.contentLength = len(outrec.contentData)
  634. self.writeRecord(outrec)
  635. def _do_begin_request(self, inrec):
  636. """Handle an FCGI_BEGIN_REQUEST from the web server."""
  637. role, flags = struct.unpack(FCGI_BeginRequestBody, inrec.contentData)
  638. req = self.server.request_class(self, self._inputStreamClass)
  639. req.requestId, req.role, req.flags = inrec.requestId, role, flags
  640. req.aborted = False
  641. if not self._multiplexed and self._requests:
  642. # Can't multiplex requests.
  643. self.end_request(req, 0L, FCGI_CANT_MPX_CONN, remove=False)
  644. else:
  645. self._requests[inrec.requestId] = req
  646. def _do_abort_request(self, inrec):
  647. """
  648. Handle an FCGI_ABORT_REQUEST from the web server.
  649. We just mark a flag in the associated Request.
  650. """
  651. req = self._requests.get(inrec.requestId)
  652. if req is not None:
  653. req.aborted = True
  654. def _start_request(self, req):
  655. """Run the request."""
  656. # Not multiplexed, so run it inline.
  657. req.run()
  658. def _do_params(self, inrec):
  659. """
  660. Handle an FCGI_PARAMS Record.
  661. If the last FCGI_PARAMS Record is received, start the request.
  662. """
  663. req = self._requests.get(inrec.requestId)
  664. if req is not None:
  665. if inrec.contentLength:
  666. pos = 0
  667. while pos < inrec.contentLength:
  668. pos, (name, value) = decode_pair(inrec.contentData, pos)
  669. req.params[name] = value
  670. else:
  671. self._start_request(req)
  672. def _do_stdin(self, inrec):
  673. """Handle the FCGI_STDIN stream."""
  674. req = self._requests.get(inrec.requestId)
  675. if req is not None:
  676. req.stdin.add_data(inrec.contentData)
  677. def _do_data(self, inrec):
  678. """Handle the FCGI_DATA stream."""
  679. req = self._requests.get(inrec.requestId)
  680. if req is not None:
  681. req.data.add_data(inrec.contentData)
  682. def _do_unknown_type(self, inrec):
  683. """Handle an unknown request type. Respond accordingly."""
  684. outrec = Record(FCGI_UNKNOWN_TYPE)
  685. outrec.contentData = struct.pack(FCGI_UnknownTypeBody, inrec.type)
  686. outrec.contentLength = FCGI_UnknownTypeBody_LEN
  687. self.writeRecord(rec)
  688. class MultiplexedConnection(Connection):
  689. """
  690. A version of Connection capable of handling multiple requests
  691. simultaneously.
  692. """
  693. _multiplexed = True
  694. _inputStreamClass = MultiplexedInputStream
  695. def __init__(self, sock, addr, server):
  696. super(MultiplexedConnection, self).__init__(sock, addr, server)
  697. # Used to arbitrate access to self._requests.
  698. lock = threading.RLock()
  699. # Notification is posted everytime a request completes, allowing us
  700. # to quit cleanly.
  701. self._lock = threading.Condition(lock)
  702. def _cleanupSocket(self):
  703. # Wait for any outstanding requests before closing the socket.
  704. self._lock.acquire()
  705. while self._requests:
  706. self._lock.wait()
  707. self._lock.release()
  708. super(MultiplexedConnection, self)._cleanupSocket()
  709. def writeRecord(self, rec):
  710. # Must use locking to prevent intermingling of Records from different
  711. # threads.
  712. self._lock.acquire()
  713. try:
  714. # Probably faster than calling super. ;)
  715. rec.write(self._sock)
  716. finally:
  717. self._lock.release()
  718. def end_request(self, req, appStatus=0L,
  719. protocolStatus=FCGI_REQUEST_COMPLETE, remove=True):
  720. self._lock.acquire()
  721. try:
  722. super(MultiplexedConnection, self).end_request(req, appStatus,
  723. protocolStatus,
  724. remove)
  725. self._lock.notify()
  726. finally:
  727. self._lock.release()
  728. def _do_begin_request(self, inrec):
  729. self._lock.acquire()
  730. try:
  731. super(MultiplexedConnection, self)._do_begin_request(inrec)
  732. finally:
  733. self._lock.release()
  734. def _do_abort_request(self, inrec):
  735. self._lock.acquire()
  736. try:
  737. super(MultiplexedConnection, self)._do_abort_request(inrec)
  738. finally:
  739. self._lock.release()
  740. def _start_request(self, req):
  741. thread.start_new_thread(req.run, ())
  742. def _do_params(self, inrec):
  743. self._lock.acquire()
  744. try:
  745. super(MultiplexedConnection, self)._do_params(inrec)
  746. finally:
  747. self._lock.release()
  748. def _do_stdin(self, inrec):
  749. self._lock.acquire()
  750. try:
  751. super(MultiplexedConnection, self)._do_stdin(inrec)
  752. finally:
  753. self._lock.release()
  754. def _do_data(self, inrec):
  755. self._lock.acquire()
  756. try:
  757. super(MultiplexedConnection, self)._do_data(inrec)
  758. finally:
  759. self._lock.release()
  760. class Server(object):
  761. """
  762. The FastCGI server.
  763. Waits for connections from the web server, processing each
  764. request.
  765. If run in a normal CGI context, it will instead instantiate a
  766. CGIRequest and run the handler through there.
  767. """
  768. request_class = Request
  769. cgirequest_class = CGIRequest
  770. # Limits the size of the InputStream's string buffer to this size + the
  771. # server's maximum Record size. Since the InputStream is not seekable,
  772. # we throw away already-read data once this certain amount has been read.
  773. inputStreamShrinkThreshold = 102400 - 8192
  774. def __init__(self, handler=None, maxwrite=8192, bindAddress=None,
  775. umask=None, multiplexed=False):
  776. """
  777. handler, if present, must reference a function or method that
  778. takes one argument: a Request object. If handler is not
  779. specified at creation time, Server *must* be subclassed.
  780. (The handler method below is abstract.)
  781. maxwrite is the maximum number of bytes (per Record) to write
  782. to the server. I've noticed mod_fastcgi has a relatively small
  783. receive buffer (8K or so).
  784. bindAddress, if present, must either be a string or a 2-tuple. If
  785. present, run() will open its own listening socket. You would use
  786. this if you wanted to run your application as an 'external' FastCGI
  787. app. (i.e. the webserver would no longer be responsible for starting
  788. your app) If a string, it will be interpreted as a filename and a UNIX
  789. socket will be opened. If a tuple, the first element, a string,
  790. is the interface name/IP to bind to, and the second element (an int)
  791. is the port number.
  792. Set multiplexed to True if you want to handle multiple requests
  793. per connection. Some FastCGI backends (namely mod_fastcgi) don't
  794. multiplex requests at all, so by default this is off (which saves
  795. on thread creation/locking overhead). If threads aren't available,
  796. this keyword is ignored; it's not possible to multiplex requests
  797. at all.
  798. """
  799. if handler is not None:
  800. self.handler = handler
  801. self.maxwrite = maxwrite
  802. if thread_available:
  803. try:
  804. import resource
  805. # Attempt to glean the maximum number of connections
  806. # from the OS.
  807. maxConns = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
  808. except ImportError:
  809. maxConns = 100 # Just some made up number.
  810. maxReqs = maxConns
  811. if multiplexed:
  812. self._connectionClass = MultiplexedConnection
  813. maxReqs *= 5 # Another made up number.
  814. else:
  815. self._connectionClass = Connection
  816. self.capability = {
  817. FCGI_MAX_CONNS: maxConns,
  818. FCGI_MAX_REQS: maxReqs,
  819. FCGI_MPXS_CONNS: multiplexed and 1 or 0
  820. }
  821. else:
  822. self._connectionClass = Connection
  823. self.capability = {
  824. # If threads aren't available, these are pretty much correct.
  825. FCGI_MAX_CONNS: 1,
  826. FCGI_MAX_REQS: 1,
  827. FCGI_MPXS_CONNS: 0
  828. }
  829. self._bindAddress = bindAddress
  830. self._umask = umask
  831. def _setupSocket(self):
  832. if self._bindAddress is None: # Run as a normal FastCGI?
  833. isFCGI = True
  834. sock = socket.fromfd(FCGI_LISTENSOCK_FILENO, socket.AF_INET,
  835. socket.SOCK_STREAM)
  836. try:
  837. sock.getpeername()
  838. except socket.error, e:
  839. if e[0] == errno.ENOTSOCK:
  840. # Not a socket, assume CGI context.
  841. isFCGI = False
  842. elif e[0] != errno.ENOTCONN:
  843. raise
  844. # FastCGI/CGI discrimination is broken on Mac OS X.
  845. # Set the environment variable FCGI_FORCE_CGI to "Y" or "y"
  846. # if you want to run your app as a simple CGI. (You can do
  847. # this with Apache's mod_env [not loaded by default in OS X
  848. # client, ha ha] and the SetEnv directive.)
  849. if not isFCGI or \
  850. os.environ.get('FCGI_FORCE_CGI', 'N').upper().startswith('Y'):
  851. req = self.cgirequest_class(self)
  852. req.run()
  853. sys.exit(0)
  854. else:
  855. # Run as a server
  856. oldUmask = None
  857. if type(self._bindAddress) is str:
  858. # Unix socket
  859. sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  860. try:
  861. os.unlink(self._bindAddress)
  862. except OSError:
  863. pass
  864. if self._umask is not None:
  865. oldUmask = os.umask(self._umask)
  866. else:
  867. # INET socket
  868. assert type(self._bindAddress) is tuple
  869. assert len(self._bindAddress) == 2
  870. sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  871. sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  872. sock.bind(self._bindAddress)
  873. sock.listen(socket.SOMAXCONN)
  874. if oldUmask is not None:
  875. os.umask(oldUmask)
  876. return sock
  877. def _cleanupSocket(self, sock):
  878. """Closes the main socket."""
  879. sock.close()
  880. def _installSignalHandlers(self):
  881. self._oldSIGs = [(x,signal.getsignal(x)) for x in
  882. (signal.SIGHUP, signal.SIGINT, signal.SIGTERM)]
  883. signal.signal(signal.SIGHUP, self._hupHandler)
  884. signal.signal(signal.SIGINT, self._intHandler)
  885. signal.signal(signal.SIGTERM, self._intHandler)
  886. def _restoreSignalHandlers(self):
  887. for signum,handler in self._oldSIGs:
  888. signal.signal(signum, handler)
  889. def _hupHandler(self, signum, frame):
  890. self._hupReceived = True
  891. self._keepGoing = False
  892. def _intHandler(self, signum, frame):
  893. self._keepGoing = False
  894. def run(self, timeout=1.0):
  895. """
  896. The main loop. Exits on SIGHUP, SIGINT, SIGTERM. Returns True if
  897. SIGHUP was received, False otherwise.
  898. """
  899. web_server_addrs = os.environ.get('FCGI_WEB_SERVER_ADDRS')
  900. if web_server_addrs is not None:
  901. web_server_addrs = map(lambda x: x.strip(),
  902. web_server_addrs.split(','))
  903. sock = self._setupSocket()
  904. self._keepGoing = True
  905. self._hupReceived = False
  906. # Install signal handlers.
  907. self._installSignalHandlers()
  908. while self._keepGoing:
  909. try:
  910. r, w, e = select.select([sock], [], [], timeout)
  911. except select.error, e:
  912. if e[0] == errno.EINTR:
  913. continue
  914. raise
  915. if r:
  916. try:
  917. clientSock, addr = sock.accept()
  918. except socket.error, e:
  919. if e[0] in (errno.EINTR, errno.EAGAIN):
  920. continue
  921. raise
  922. if web_server_addrs and \
  923. (len(addr) != 2 or addr[0] not in web_server_addrs):
  924. clientSock.close()
  925. continue
  926. # Instantiate a new Connection and begin processing FastCGI
  927. # messages (either in a new thread or this thread).
  928. conn = self._connectionClass(clientSock, addr, self)
  929. thread.start_new_thread(conn.run, ())
  930. self._mainloopPeriodic()
  931. # Restore signal handlers.
  932. self._restoreSignalHandlers()
  933. self._cleanupSocket(sock)
  934. return self._hupReceived
  935. def _mainloopPeriodic(self):
  936. """
  937. Called with just about each iteration of the main loop. Meant to
  938. be overridden.
  939. """
  940. pass
  941. def _exit(self, reload=False):
  942. """
  943. Protected convenience method for subclasses to force an exit. Not
  944. really thread-safe, which is why it isn't public.
  945. """
  946. if self._keepGoing:
  947. self._keepGoing = False
  948. self._hupReceived = reload
  949. def handler(self, req):
  950. """
  951. Default handler, which just raises an exception. Unless a handler
  952. is passed at initialization time, this must be implemented by
  953. a subclass.
  954. """
  955. raise NotImplementedError, self.__class__.__name__ + '.handler'
  956. def error(self, req):
  957. """
  958. Called by Request if an exception occurs within the handler. May and
  959. should be overridden.
  960. """
  961. if config.DEBUG:
  962. import cgitb
  963. req.stdout.write('Content-Type: text/html\r\n\r\n' +
  964. cgitb.html(sys.exc_info()))
  965. else:
  966. req.stdout.write('Content-Type: text/html\r\n\r\n' +
  967. 'Internal server error.')
  968. class WSGIServer(Server):
  969. """
  970. FastCGI server that supports the Web Server Gateway Interface. See
  971. <http://www.python.org/peps/pep-0333.html>.
  972. """
  973. def __init__(self, application, environ=None, umask=None,
  974. multithreaded=True, **kw):
  975. """
  976. environ, if present, must be a dictionary-like object. Its
  977. contents will be copied into application's environ. Useful
  978. for passing application-specific variables.
  979. Set multithreaded to False if your application is not MT-safe.
  980. """
  981. if kw.has_key('handler'):
  982. del kw['handler'] # Doesn't make sense to let this through
  983. super(WSGIServer, self).__init__(**kw)
  984. if environ is None:
  985. environ = {}
  986. self.application = application
  987. self.environ = environ
  988. self.multithreaded = multithreaded
  989. # Used to force single-threadedness
  990. self._app_lock = thread.allocate_lock()
  991. def handler(self, req):
  992. """Special handler for WSGI."""
  993. if req.role != FCGI_RESPONDER:
  994. return FCGI_UNKNOWN_ROLE, 0
  995. # Mostly taken from example CGI gateway.
  996. environ = req.params
  997. environ.update(self.environ)
  998. environ['wsgi.version'] = (1,0)
  999. environ['wsgi.input'] = req.stdin
  1000. if self._bindAddress is None:
  1001. stderr = req.stderr
  1002. else:
  1003. stderr = TeeOutputStream((sys.stderr, req.stderr))
  1004. environ['wsgi.errors'] = stderr
  1005. environ['wsgi.multithread'] = not isinstance(req, CGIRequest) and \
  1006. thread_available and self.multithreaded
  1007. # Rationale for the following: If started by the web server
  1008. # (self._bindAddress is None) in either FastCGI or CGI mode, the
  1009. # possibility of being spawned multiple times simultaneously is quite
  1010. # real. And, if started as an external server, multiple copies may be
  1011. # spawned for load-balancing/redundancy. (Though I don't think
  1012. # mod_fastcgi supports this?)
  1013. environ['wsgi.multiprocess'] = True
  1014. environ['wsgi.run_once'] = isinstance(req, CGIRequest)
  1015. if environ.get('HTTPS', 'off') in ('on', '1'):
  1016. environ['wsgi.url_scheme'] = 'https'
  1017. else:
  1018. environ['wsgi.url_scheme'] = 'http'
  1019. self._sanitizeEnv(environ)
  1020. headers_set = []
  1021. headers_sent = []
  1022. result = None
  1023. def write(data):
  1024. assert type(data) is str, 'write() argument must be string'
  1025. assert headers_set, 'write() before start_response()'
  1026. if not headers_sent:
  1027. status, responseHeaders = headers_sent[:] = headers_set
  1028. found = False
  1029. for header,value in responseHeaders:
  1030. if header.lower() == 'content-length':
  1031. found = True
  1032. break
  1033. if not found and result is not None:
  1034. try:
  1035. if len(result) == 1:
  1036. responseHeaders.append(('Content-Length',
  1037. str(len(data))))
  1038. except:
  1039. pass
  1040. s = 'Status: %s\r\n' % status
  1041. for header in responseHeaders:
  1042. s += '%s: %s\r\n' % header
  1043. s += '\r\n'
  1044. req.stdout.write(s)
  1045. req.stdout.write(data)
  1046. req.stdout.flush()
  1047. def start_response(status, response_headers, exc_info=None):
  1048. if exc_info:
  1049. try:
  1050. if headers_sent:
  1051. # Re-raise if too late
  1052. raise exc_info[0], exc_info[1], exc_info[2]
  1053. finally:
  1054. exc_info = None # avoid dangling circular ref
  1055. else:
  1056. assert not headers_set, 'Headers already set!'
  1057. assert type(status) is str, 'Status must be a string'
  1058. assert len(status) >= 4, 'Status must be at least 4 characters'
  1059. assert int(status[:3]), 'Status must begin with 3-digit code'
  1060. assert status[3] == ' ', 'Status must have a space after code'
  1061. assert type(response_headers) is list, 'Headers must be a list'
  1062. if __debug__:
  1063. for name,val in response_headers:
  1064. assert type(name) is str, 'Header names must be strings'
  1065. assert type(val) is str, 'Header values must be strings'
  1066. headers_set[:] = [status, response_headers]
  1067. return write
  1068. if not self.multithreaded:
  1069. self._app_lock.acquire()
  1070. try:
  1071. try:
  1072. result = self.application(environ, start_response)
  1073. try:
  1074. for data in result:
  1075. if data:
  1076. write(data)
  1077. if not headers_sent:
  1078. write('') # in case body was empty
  1079. finally:
  1080. if hasattr(result, 'close'):
  1081. result.close()
  1082. except socket.error, e:
  1083. if e[0] != errno.EPIPE:
  1084. raise # Don't let EPIPE propagate beyond server
  1085. finally:
  1086. if not self.multithreaded:
  1087. self._app_lock.release()
  1088. return FCGI_REQUEST_COMPLETE, 0
  1089. def _sanitizeEnv(self, environ):
  1090. """Ensure certain values are present, if required by WSGI."""
  1091. if not environ.has_key('SCRIPT_NAME'):
  1092. environ['SCRIPT_NAME'] = ''
  1093. if not environ.has_key('PATH_INFO'):
  1094. environ['PATH_INFO'] = ''
  1095. # If any of these are missing, it probably signifies a broken
  1096. # server...
  1097. for name,default in [('REQUEST_METHOD', 'GET'),
  1098. ('SERVER_NAME', 'localhost'),
  1099. ('SERVER_PORT', '80'),
  1100. ('SERVER_PROTOCOL', 'HTTP/1.0')]:
  1101. if not environ.has_key(name):
  1102. environ['wsgi.errors'].write('%s: missing FastCGI param %s '
  1103. 'required by WSGI!\n' %
  1104. (self.__class__.__name__, name))
  1105. environ[name] = default
  1106. if __name__ == '__main__':
  1107. def test_app(environ, start_response):
  1108. """Probably not the most efficient example."""
  1109. import cgi
  1110. start_response('200 OK', [('Content-Type', 'text/html')])
  1111. yield '<html><head><title>Hello World!</title></head>\n' \
  1112. '<body>\n' \
  1113. '<p>Hello World!</p>\n' \
  1114. '<table border="1">'
  1115. names = environ.keys()
  1116. names.sort()
  1117. for name in names:
  1118. yield '<tr><td>%s</td><td>%s</td></tr>\n' % (
  1119. name, cgi.escape(`environ[name]`))
  1120. form = cgi.FieldStorage(fp=environ['wsgi.input'], environ=environ,
  1121. keep_blank_values=1)
  1122. if form.list:
  1123. yield '<tr><th colspan="2">Form data</th></tr>'
  1124. for field in form.list:
  1125. yield '<tr><td>%s</td><td>%s</td></tr>\n' % (
  1126. field.name, field.value)
  1127. yield '</table>\n' \
  1128. '</body></html>\n'
  1129. WSGIServer(test_app).run()