diff options
| author | Jimmy Huang <jimmy.huang@linux.intel.com> | 2012-08-31 15:24:14 -0700 |
|---|---|---|
| committer | Jimmy Huang <jimmy.huang@linux.intel.com> | 2012-08-31 15:24:14 -0700 |
| commit | f007752892d78fb2d2cddbf5dd2f8a61574d91ed (patch) | |
| tree | fe359b1f287d6780700456937c25bf0df8a33995 /autobahn | |
| download | python-autobahn-2.0.tar.gz python-autobahn-2.0.tar.bz2 python-autobahn-2.0.zip | |
Initial import to TizenHEADsubmit/trunk/20120831.222423submit/2.0alpha/20121116.194117submit/2.0/20130320.030059accepted/trunk/20120904.170034accepted/2.0alpha/20121116.234516accepted/2.0/20130320.0404041.0_branch2.0alpha2.01.0
Signed-off-by: Jimmy Huang <jimmy.huang@linux.intel.com>
Diffstat (limited to 'autobahn')
| -rw-r--r-- | autobahn/__init__.py | 33 | ||||
| -rw-r--r-- | autobahn/httpstatus.py | 271 | ||||
| -rw-r--r-- | autobahn/prefixmap.py | 142 | ||||
| -rw-r--r-- | autobahn/useragent.py | 309 | ||||
| -rw-r--r-- | autobahn/utf8validator.py | 111 | ||||
| -rw-r--r-- | autobahn/util.py | 60 | ||||
| -rw-r--r-- | autobahn/wamp.py | 1719 | ||||
| -rw-r--r-- | autobahn/websocket.py | 3636 | ||||
| -rw-r--r-- | autobahn/xormasker.py | 85 |
9 files changed, 6366 insertions, 0 deletions
diff --git a/autobahn/__init__.py b/autobahn/__init__.py new file mode 100644 index 0000000..673fcdd --- /dev/null +++ b/autobahn/__init__.py @@ -0,0 +1,33 @@ +###############################################################################
+##
+## Copyright 2011 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+try:
+ import pkg_resources
+ version = pkg_resources.require("Autobahn")[0].version
+except:
+ ## i.e. no setuptools or no package installed ..
+ version = "?.?.?"
+
+import util
+import useragent
+import httpstatus
+import utf8validator
+import xormasker
+import websocket
+import prefixmap
+import wamp
diff --git a/autobahn/httpstatus.py b/autobahn/httpstatus.py new file mode 100644 index 0000000..9c8aa03 --- /dev/null +++ b/autobahn/httpstatus.py @@ -0,0 +1,271 @@ +###############################################################################
+##
+## Copyright 2011 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+##
+## HTTP Status Codes
+##
+## Source: http://en.wikipedia.org/wiki/List_of_HTTP_status_codes
+## Adapted on 2011/10/11
+##
+
+##
+## 1xx Informational
+##
+## Request received, continuing process.
+##
+## This class of status code indicates a provisional response, consisting only of
+## the Status-Line and optional headers, and is terminated by an empty line.
+## Since HTTP/1.0 did not define any 1xx status codes, servers must not send
+## a 1xx response to an HTTP/1.0 client except under experimental conditions.
+##
+
+HTTP_STATUS_CODE_CONTINUE = (100, "Continue",
+ "This means that the server has received the request headers, and that the client should proceed to send the request body (in the case of a request for which a body needs to be sent; for example, a POST request). If the request body is large, sending it to a server when a request has already been rejected based upon inappropriate headers is inefficient. To have a server check if the request could be accepted based on the request's headers alone, a client must send Expect: 100-continue as a header in its initial request[2] and check if a 100 Continue status code is received in response before continuing (or receive 417 Expectation Failed and not continue).")
+
+HTTP_STATUS_CODE_SWITCHING_PROTOCOLS = (101, "Switching Protocols",
+ "This means the requester has asked the server to switch protocols and the server is acknowledging that it will do so.")
+
+HTTP_STATUS_CODE_PROCESSING = (102, "Processing (WebDAV) (RFC 2518)",
+ "As a WebDAV request may contain many sub-requests involving file operations, it may take a long time to complete the request. This code indicates that the server has received and is processing the request, but no response is available yet.[3] This prevents the client from timing out and assuming the request was lost.")
+
+HTTP_STATUS_CODE_CHECKPOINT = (103, "Checkpoint",
+ "This code is used in the Resumable HTTP Requests Proposal to resume aborted PUT or POST requests.")
+
+HTTP_STATUS_CODE_REQUEST_URI_TOO_LONG = (122, "Request-URI too long",
+ "This is a non-standard IE7-only code which means the URI is longer than a maximum of 2083 characters.[5][6] (See code 414.)")
+
+##
+## 2xx Success
+##
+## This class of status codes indicates the action requested by the client was
+## received, understood, accepted and processed successfully.
+##
+
+HTTP_STATUS_CODE_OK = (200, "OK",
+ "Standard response for successful HTTP requests. The actual response will depend on the request method used. In a GET request, the response will contain an entity corresponding to the requested resource. In a POST request the response will contain an entity describing or containing the result of the action.")
+
+HTTP_STATUS_CODE_CREATED = (201, "Created",
+ "The request has been fulfilled and resulted in a new resource being created.")
+
+HTTP_STATUS_CODE_ACCEPTED = (202, "Accepted",
+ "The request has been accepted for processing, but the processing has not been completed. The request might or might not eventually be acted upon, as it might be disallowed when processing actually takes place.")
+
+HTTP_STATUS_CODE_NON_AUTHORATIVE = (203, "Non-Authoritative Information (since HTTP/1.1)",
+ "The server successfully processed the request, but is returning information that may be from another source.")
+
+HTTP_STATUS_CODE_NO_CONTENT = (204, "No Content",
+ "The server successfully processed the request, but is not returning any content.")
+
+HTTP_STATUS_CODE_RESET_CONTENT = (205, "Reset Content",
+ "The server successfully processed the request, but is not returning any content. Unlike a 204 response, this response requires that the requester reset the document view.")
+
+HTTP_STATUS_CODE_PARTIAL_CONTENT = (206, "Partial Content",
+ "The server is delivering only part of the resource due to a range header sent by the client. The range header is used by tools like wget to enable resuming of interrupted downloads, or split a download into multiple simultaneous streams.")
+
+HTTP_STATUS_CODE_MULTI_STATUS = (207, "Multi-Status (WebDAV) (RFC 4918)",
+ "The message body that follows is an XML message and can contain a number of separate response codes, depending on how many sub-requests were made.")
+
+HTTP_STATUS_CODE_IM_USED = (226, "IM Used (RFC 3229)",
+ "The server has fulfilled a GET request for the resource, and the response is a representation of the result of one or more instance-manipulations applied to the current instance.")
+
+##
+## 3xx Redirection
+##
+## The client must take additional action to complete the request.
+##
+## This class of status code indicates that further action needs to be taken
+## by the user agent in order to fulfil the request. The action required may
+## be carried out by the user agent without interaction with the user if and
+## only if the method used in the second request is GET or HEAD. A user agent
+## should not automatically redirect a request more than five times, since such
+## redirections usually indicate an infinite loop.
+##
+
+HTTP_STATUS_CODE_MULTIPLE_CHOICES = (300, "Multiple Choices",
+ "Indicates multiple options for the resource that the client may follow. It, for instance, could be used to present different format options for video, list files with different extensions, or word sense disambiguation.")
+
+HTTP_STATUS_CODE_MOVED_PERMANENTLY = (301, "Moved Permanently",
+ "This and all future requests should be directed to the given URI.")
+
+HTTP_STATUS_CODE_FOUND = (302, "Found",
+ "This is an example of industrial practice contradicting the standard. HTTP/1.0 specification (RFC 1945) required the client to perform a temporary redirect (the original describing phrase was 'Moved Temporarily', but popular browsers implemented 302 with the functionality of a 303 See Other. Therefore, HTTP/1.1 added status codes 303 and 307 to distinguish between the two behaviours. However, some Web applications and frameworks use the 302 status code as if it were the 303.")
+
+HTTP_STATUS_CODE_SEE_OTHER = (303, "See Other (since HTTP/1.1)",
+ "The response to the request can be found under another URI using a GET method. When received in response to a POST (or PUT/DELETE), it should be assumed that the server has received the data and the redirect should be issued with a separate GET message.")
+
+HTTP_STATUS_CODE_NOT_MODIFIED = (304, "Not Modified",
+ "Indicates the resource has not been modified since last requested.[2] Typically, the HTTP client provides a header like the If-Modified-Since header to provide a time against which to compare. Using this saves bandwidth and reprocessing on both the server and client, as only the header data must be sent and received in comparison to the entirety of the page being re-processed by the server, then sent again using more bandwidth of the server and client.")
+
+HTTP_STATUS_CODE_USE_PROXY = (305, "Use Proxy (since HTTP/1.1)",
+ "Many HTTP clients (such as Mozilla[11] and Internet Explorer) do not correctly handle responses with this status code, primarily for security reasons.")
+
+HTTP_STATUS_CODE_SWITCH_PROXY = (306, "Switch Proxy",
+ "No longer used. Originally meant 'Subsequent requests should use the specified proxy'.")
+
+HTTP_STATUS_CODE_TEMPORARY_REDIRECT = (307, "Temporary Redirect (since HTTP/1.1)",
+ "In this occasion, the request should be repeated with another URI, but future requests can still use the original URI.[2] In contrast to 303, the request method should not be changed when reissuing the original request. For instance, a POST request must be repeated using another POST request.")
+
+HTTP_STATUS_CODE_RESUME_INCOMPLETE = (308, "Resume Incomplete",
+ "This code is used in the Resumable HTTP Requests Proposal to resume aborted PUT or POST requests.")
+
+##
+## 4xx Client Error
+##
+## The 4xx class of status code is intended for cases in which the client
+## seems to have erred. Except when responding to a HEAD request, the server
+## should include an entity containing an explanation of the error situation,
+## and whether it is a temporary or permanent condition. These status codes are
+## applicable to any request method. User agents should display any included
+## entity to the user. These are typically the most common error codes
+## encountered while online.
+##
+
+HTTP_STATUS_CODE_BAD_REQUEST = (400, "Bad Request",
+ "The request cannot be fulfilled due to bad syntax.")
+
+HTTP_STATUS_CODE_UNAUTHORIZED = (401, "Unauthorized",
+ "Similar to 403 Forbidden, but specifically for use when authentication is possible but has failed or not yet been provided.[2] The response must include a WWW-Authenticate header field containing a challenge applicable to the requested resource. See Basic access authentication and Digest access authentication.")
+
+HTTP_STATUS_CODE_PAYMENT_REQUIRED = (402, "Payment Required",
+ "Reserved for future use.[2] The original intention was that this code might be used as part of some form of digital cash or micropayment scheme, but that has not happened, and this code is not usually used. As an example of its use, however, Apple's MobileMe service generates a 402 error if the MobileMe account is delinquent.")
+
+HTTP_STATUS_CODE_FORBIDDEN = (403, "Forbidden",
+ "The request was a legal request, but the server is refusing to respond to it.[2] Unlike a 401 Unauthorized response, authenticating will make no difference.[2]")
+
+HTTP_STATUS_CODE_NOT_FOUND = (404, "Not Found",
+ "The requested resource could not be found but may be available again in the future.[2] Subsequent requests by the client are permissible.")
+
+HTTP_STATUS_CODE_METHOD_NOT_ALLOWED = (405, "Method Not Allowed",
+ "A request was made of a resource using a request method not supported by that resource;[2] for example, using GET on a form which requires data to be presented via POST, or using PUT on a read-only resource.")
+
+HTTP_STATUS_CODE_NOT_ACCEPTABLE = (406, "Not Acceptable",
+ "The requested resource is only capable of generating content not acceptable according to the Accept headers sent in the request.")
+
+HTTP_STATUS_CODE_PROXY_AUTH_REQUIRED = (407, "Proxy Authentication Required",
+ "The client must first authenticate itself with the proxy.")
+
+HTTP_STATUS_CODE_REQUEST_TIMEOUT = (408, "Request Timeout",
+ "The server timed out waiting for the request. According to W3 HTTP specifications: 'The client did not produce a request within the time that the server was prepared to wait. The client MAY repeat the request without modifications at any later time.'")
+
+HTTP_STATUS_CODE_CONFLICT = (409, "Conflict",
+ "Indicates that the request could not be processed because of conflict in the request, such as an edit conflict.")
+
+HTTP_STATUS_CODE_GONE = (410, "Gone",
+ "Indicates that the resource requested is no longer available and will not be available again.[2] This should be used when a resource has been intentionally removed and the resource should be purged. Upon receiving a 410 status code, the client should not request the resource again in the future. Clients such as search engines should remove the resource from their indices. Most use cases do not require clients and search engines to purge the resource, and a '404 Not Found' may be used instead.")
+
+HTTP_STATUS_CODE_LENGTH_REQUIRED = (411, "Length Required",
+ "The request did not specify the length of its content, which is required by the requested resource.")
+
+HTTP_STATUS_CODE_PRECONDITION_FAILED = (412, "Precondition Failed",
+ "The server does not meet one of the preconditions that the requester put on the request.")
+
+HTTP_STATUS_CODE_REQUEST_ENTITY_TOO_LARGE = (413, "Request Entity Too Large",
+ "The request is larger than the server is willing or able to process.")
+
+HTTP_STATUS_CODE_REQUEST_URI_TOO_LARGE = (414, "Request-URI Too Long",
+ "The URI provided was too long for the server to process.")
+
+HTTP_STATUS_CODE_UNSUPPORTED_MEDIA_TYPE = (415, "Unsupported Media Type",
+ "The request entity has a media type which the server or resource does not support. For example, the client uploads an image as image/svg+xml, but the server requires that images use a different format.")
+
+HTTP_STATUS_CODE_INVALID_REQUEST_RANGE = (416, "Requested Range Not Satisfiable",
+ "The client has asked for a portion of the file, but the server cannot supply that portion.[2] For example, if the client asked for a part of the file that lies beyond the end of the file.")
+
+HTTP_STATUS_CODE_EXPECTATION_FAILED = (417, "Expectation Failed",
+ "The server cannot meet the requirements of the Expect request-header field.")
+
+HTTP_STATUS_CODE_TEAPOT = (418, "I'm a teapot (RFC 2324)",
+ "This code was defined in 1998 as one of the traditional IETF April Fools' jokes, in RFC 2324, Hyper Text Coffee Pot Control Protocol, and is not expected to be implemented by actual HTTP servers.")
+
+HTTP_STATUS_CODE_UNPROCESSABLE_ENTITY = (422, "Unprocessable Entity (WebDAV) (RFC 4918)",
+ "The request was well-formed but was unable to be followed due to semantic errors.")
+
+HTTP_STATUS_CODE_LOCKED = (423, "Locked (WebDAV) (RFC 4918)",
+ "The resource that is being accessed is locked.")
+
+HTTP_STATUS_CODE_FAILED_DEPENDENCY = (424, "Failed Dependency (WebDAV) (RFC 4918)",
+ "The request failed due to failure of a previous request (e.g. a PROPPATCH).")
+
+HTTP_STATUS_CODE_UNORDERED_COLLECTION = (425, "Unordered Collection (RFC 3648)",
+ "Defined in drafts of 'WebDAV Advanced Collections Protocol', but not present in 'Web Distributed Authoring and Versioning (WebDAV) Ordered Collections Protocol'.")
+
+HTTP_STATUS_CODE_UPGRADE_REQUIRED = (426, "Upgrade Required (RFC 2817)",
+ "The client should switch to a different protocol such as TLS/1.0.")
+
+HTTP_STATUS_CODE_NO_RESPONSE = (444, "No Response",
+ "A Nginx HTTP server extension. The server returns no information to the client and closes the connection (useful as a deterrent for malware).")
+
+HTTP_STATUS_CODE_RETRY_WITH = (449, "Retry With",
+ "A Microsoft extension. The request should be retried after performing the appropriate action.")
+
+HTTP_STATUS_CODE_PARANTAL_BLOCKED = (450, "Blocked by Windows Parental Controls",
+ "A Microsoft extension. This error is given when Windows Parental Controls are turned on and are blocking access to the given webpage.")
+
+HTTP_STATUS_CODE_CLIENT_CLOSED_REQUEST = (499, "Client Closed Request",
+ "An Nginx HTTP server extension. This code is introduced to log the case when the connection is closed by client while HTTP server is processing its request, making server unable to send the HTTP header back.")
+
+
+##
+## 5xx Server Error
+##
+## The server failed to fulfill an apparently valid request.
+##
+## Response status codes beginning with the digit "5" indicate cases in which
+## the server is aware that it has encountered an error or is otherwise incapable
+## of performing the request. Except when responding to a HEAD request, the server
+## should include an entity containing an explanation of the error situation, and
+## indicate whether it is a temporary or permanent condition. Likewise, user agents
+## should display any included entity to the user. These response codes are
+## applicable to any request method.
+##
+
+HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR = (500, "Internal Server Error",
+ "A generic error message, given when no more specific message is suitable.")
+
+HTTP_STATUS_CODE_NOT_IMPLEMENTED = (501, "Not Implemented",
+ "The server either does not recognise the request method, or it lacks the ability to fulfill the request.")
+
+HTTP_STATUS_CODE_BAD_GATEWAY = (502, "Bad Gateway",
+ "The server was acting as a gateway or proxy and received an invalid response from the upstream server.")
+
+HTTP_STATUS_CODE_SERVICE_UNAVAILABLE = (503, "Service Unavailable",
+ "The server is currently unavailable (because it is overloaded or down for maintenance). Generally, this is a temporary state.")
+
+HTTP_STATUS_CODE_GATEWAY_TIMEOUT = (504, "Gateway Timeout",
+ "The server was acting as a gateway or proxy and did not receive a timely response from the upstream server.")
+
+HTTP_STATUS_CODE_UNSUPPORTED_HTTP_VERSION = (505, "HTTP Version Not Supported",
+ "The server does not support the HTTP protocol version used in the request.")
+
+HTTP_STATUS_CODE_VARIANT_ALSO_NEGOTIATES = (506, "Variant Also Negotiates (RFC 2295)",
+ "Transparent content negotiation for the request results in a circular reference.")
+
+HTTP_STATUS_CODE_INSUFFICIENT_STORAGE = (507, "Insufficient Storage (WebDAV)(RFC 4918)",
+ "The server is unable to store the representation needed to complete the request.")
+
+HTTP_STATUS_CODE_BANDWIDTH_LIMIT_EXCEEDED = (509, "Bandwidth Limit Exceeded (Apache bw/limited extension)",
+ "This status code, while used by many servers, is not specified in any RFCs.")
+
+HTTP_STATUS_CODE_NOT_EXTENDED = (510, "Not Extended (RFC 2774)",
+ "Further extensions to the request are required for the server to fulfill it.")
+
+HTTP_STATUS_CODE_NETWORK_READ_TIMEOUT = (598, "Network read timeout error (Informal convention)",
+ "This status code is not specified in any RFCs, but is used by some HTTP proxies to signal a network read timeout behind the proxy to a client in front of the proxy.")
+
+HTTP_STATUS_CODE_NETWORK_CONNECT_TIMEOUT = (599, "Network connect timeout error (Informal convention)",
+ "This status code is not specified in any RFCs, but is used by some HTTP proxies to signal a network connect timeout behind the proxy to a client in front of the proxy.")
diff --git a/autobahn/prefixmap.py b/autobahn/prefixmap.py new file mode 100644 index 0000000..2a1198f --- /dev/null +++ b/autobahn/prefixmap.py @@ -0,0 +1,142 @@ +###############################################################################
+##
+## Copyright 2011 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+class PrefixMap:
+ """
+ Provides a two-way mapping between CURIEs (Compact URI Expressions) and
+ full URIs. See http://www.w3.org/TR/curie/.
+ """
+
+ def __init__(self):
+ self.index = {}
+ self.rindex = {}
+
+ ## add a couple of well-know prefixes
+ ##
+ #self.set("owl", "http://www.w3.org/2002/07/owl#")
+ #self.set("rdf", "http://www.w3.org/1999/02/22-rdf-syntax-ns#")
+ #self.set("rdfs", "http://www.w3.org/2000/01/rdf-schema#")
+ #self.set("rdfa", "http://www.w3.org/ns/rdfa#")
+ #self.set("xhv", "http://www.w3.org/1999/xhtml/vocab#")
+ #self.set("xml", "http://www.w3.org/XML/1998/namespace")
+ #self.set("xsd", "http://www.w3.org/2001/XMLSchema#")
+
+
+ def get(self, prefix):
+ """
+ Returns the URI for the prefix or None if prefix has no mapped URI.
+
+ :param prefix: Prefix to map.
+ :type prefix: str
+ :returns: str -- Mapped URI for prefix or None.
+ """
+ return self.index.get(prefix, None)
+
+
+ def set(self, prefix, uri):
+ """
+ Set mapping of prefix to URI.
+
+ :param prefix: Prefix to be mapped.
+ :type prefix: str
+ :param uri: URI the prefix is to be mapped to.
+ :type uri: str
+ """
+ self.index[prefix] = uri
+ self.rindex[uri] = prefix
+
+
+ def setDefault(self, uri):
+ """
+ Set default URI mapping of empty prefix (prefix of length 0).
+
+ :param uri: URI the empty prefix to be mapped to (i.e. :label should map to uri:label).
+ :type str
+ """
+ self.set("", uri)
+
+
+ def remove(self, prefix):
+ """
+ Remove mapping of prefix to URI.
+
+ :param prefix: Prefix for which mapping should be removed.
+ :type str
+ """
+ uri = index.get(index, None)
+ if uri:
+ del self.index[prefix]
+ del self.rindex[uri]
+
+
+ def resolve(self, curie):
+ """
+ Resolve given CURIE to full URI.
+
+ :param curie: CURIE (i.e. "rdf:label").
+ :type curie: str
+ :returns: str -- Full URI for CURIE or None.
+ """
+ i = curie.find(":")
+ if i > 0:
+ prefix = curie[:i]
+ if self.index.has_key(prefix):
+ return self.index[prefix] + curie[i+1:]
+ return None
+
+
+ def resolveOrPass(self, curieOrUri):
+ """
+ Resolve given CURIE/URI and return string verbatim if cannot be resolved.
+
+ :param curieOrUri: CURIE or URI.
+ :type curieOrUri: str
+ :returns: str -- Full URI for CURIE or original string.
+ """
+ u = self.resolve(curieOrUri)
+ if u:
+ return u
+ else:
+ return curieOrUri
+
+
+ def shrink(self, uri):
+ """
+ Shrink given URI to CURIE. If no appropriate prefix mapping is available,
+ return original URI.
+
+ :param uri: URI to shrink.
+ :type uri: str
+ :returns str -- CURIE or original URI.
+ """
+ for i in xrange(len(uri), 1, -1):
+ u = uri[:i]
+ p = self.rindex.get(u, None)
+ if p:
+ return p + ":" + uri[i:]
+ return uri
+
+
+if __name__ == '__main__':
+ m = PrefixMap()
+ print m.resolve("http://www.w3.org/1999/02/22-rdf-syntax-ns#label")
+ print m.resolve("rdf:label")
+ print m.resolve("foobar:label")
+ print m.shrink("http://www.w3.org/1999/02/22-rdf-syntax-ns#")
+ print m.shrink("http://www.w3.org/1999/02/22-rdf-syntax-ns#label")
+ print m.shrink("http://foobar.org#label")
diff --git a/autobahn/useragent.py b/autobahn/useragent.py new file mode 100644 index 0000000..83d7178 --- /dev/null +++ b/autobahn/useragent.py @@ -0,0 +1,309 @@ +###############################################################################
+##
+## Copyright 2011-2012 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+import re
+UA_FIREFOX = re.compile(".*Firefox/(\d*).*")
+UA_CHROME = re.compile(".*Chrome/(\d*).*")
+UA_CHROMEFRAME = re.compile(".*chromeframe/(\d*).*")
+UA_WEBKIT = re.compile(".*AppleWebKit/([0-9+\.]*)\w*.*")
+UA_WEBOS = re.compile(".*webos/([0-9+\.]*)\w*.*")
+UA_HPWEBOS = re.compile(".*hpwOS/([0-9+\.]*)\w*.*")
+
+
+
+# Chrome =============================================================
+
+# Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/17.0.963.56 Safari/535.11
+
+
+# Chrome Frame =======================================================
+
+# IE6 on Windows with Chrome Frame
+# Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; chromeframe/11.0.660.0)
+
+
+# Firefox ============================================================
+
+# Windows 7 64 Bit
+# Mozilla/5.0 (Windows NT 6.1; WOW64; rv:12.0a2) Gecko/20120227 Firefox/12.0a2
+
+
+# Android ============================================================
+
+# Firefox Mobile
+# Mozilla/5.0 (Android; Linux armv7l; rv:10.0.2) Gecko/20120215 Firefox/10.0.2 Fennec/10.0.2
+
+# Chrome for Android (on ICS)
+# Mozilla/5.0 (Linux; U; Android-4.0.3; en-us; Galaxy Nexus Build/IML74K) AppleWebKit/535.7 (KHTML, like Gecko) CrMo/16.0.912.75 Mobile Safari/535.7
+
+# Android builtin browser
+
+# Samsung Galaxy Tab 1
+# Mozilla/5.0 (Linux; U; Android 2.2; de-de; GT-P1000 Build/FROYO) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1
+
+# Samsung Galaxy S
+# Mozilla/5.0 (Linux; U; Android 2.3.3; de-de; GT-I9000 Build/GINGERBREAD) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1
+
+# Samsung Galaxy Note
+# Mozilla/5.0 (Linux; U; Android 2.3.6; de-de; GT-N7000 Build/GINGERBREAD) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1
+
+# Samsung Galaxy ACE (no Flash since ARM)
+# Mozilla/5.0 (Linux; U; Android 2.2.1; de-de; GT-S5830 Build/FROYO) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1
+
+
+# WebOS ==============================================================
+
+# HP Touchpad
+# Mozilla/5.0 (hp-tablet; Linux; hpwOS/3.0.5; U; en-US) AppleWebKit/534.6 (KHTML, like Gecko) wOSBrowser/234.83 Safari/534.6 TouchPad/1.0
+# => Qt-WebKit, Hixie-76, Flash
+
+
+# Safari =============================================================
+
+# iPod Touch, iOS 4.2.1
+# Mozilla/5.0 (iPod; U; CPU iPhone OS 4_2_1 like Mac OS X; de-de) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8C148 Safari/6533.18.5
+# => Hixie-76
+
+# MacBook Pro, OSX 10.5.8, Safari 5.0.6
+# Mozilla/5.0 (Macintosh; Intel Mac OS X 10_5_8) AppleWebKit/534.50.2 (KHTML, like Gecko) Version/5.0.6 Safari/533.22.3
+# => Hixie-76
+
+# RFC6455
+# Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/534+ (KHTML, like Gecko) Version/5.1.2 Safari/534.52.7
+# Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_3) AppleWebKit/535.24+ (KHTML, like Gecko) Version/5.1.3 Safari/534.53.10
+
+# Hixie-76
+# Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_3) AppleWebKit/534.53.11 (KHTML, like Gecko) Version/5.1.3 Safari/534.53.10
+
+# Hixie-76
+# Mozilla/5.0 (Macintosh; Intel Mac OS X 10_5_8) AppleWebKit/534.50.2 (KHTML, like Gecko) Version/5.0.6 Safari/533.22.3
+
+
+# Opera ==============================================================
+
+# Windows 7 32-Bit
+# Opera/9.80 (Windows NT 6.1; U; de) Presto/2.10.229 Version/11.61
+
+# Windows 7 64-Bit
+# Opera/9.80 (Windows NT 6.1; WOW64; U; de) Presto/2.10.229 Version/11.62
+
+# Samsung Galaxy S
+# Opera/9.80 (Android 2.3.3; Linux; Opera Mobi/ADR-1202231246; U; de) Presto/2.10.254 Version/12.00
+
+# Samsung Galaxy Tab 1
+# Opera/9.80 (Android 2.2; Linux; Opera Tablet/ADR-1203051631; U; de) Presto/2.10.254 Version/12.00
+
+# Samsung Galaxy ACE:
+# Opera/9.80 (Android 2.2.1; Linux; Opera Mobi/ADR-1203051631; U; de) Presto/2.10.254 Version/12.00
+
+# Nokia N8, Symbian S60 5th Ed., S60 Bell
+# Opera/9.80 (S60; SymbOS; Opera Mobi/SYB-1111151949; U; de) Presto/2.9.201 Version/11.50
+
+
+def _lookupWsSupport(ua):
+ """
+ Lookup if browser supports WebSocket (Hixie76, Hybi10+, RFC6455) natively,
+ and if not, whether the web-socket-js Flash bridge works to polyfill that.
+
+ Returns a tuple of booleans
+
+ (ws_supported, needs_flash, detected)
+
+ ws_supported = WebSocket is supported
+ needs_flash = Flash Bridge is needed for support
+ detected = the code has explicitly mapped the support/nosupport
+
+ Params:
+
+ ua = user agent string, i.e. flask.request.user_agent.string
+ """
+
+ ## Internet Explorer
+ ##
+ ## FIXME: handle Windows Phone
+ ##
+ if ua.find("MSIE") >= 0:
+ # IE10 has native support
+ if ua.find("MSIE 10") >= 0:
+ # native Hybi-10+
+ return (True, False, True)
+
+ # first, check for Google Chrome Frame
+ # http://www.chromium.org/developers/how-tos/chrome-frame-getting-started/understanding-chrome-frame-user-agent
+ if ua.find("chromeframe") >= 0:
+
+ r = UA_CHROMEFRAME.match(ua)
+ try:
+ v = int(r.groups()[0])
+ if v >= 14:
+ # native Hybi-10+
+ return (True, False, True)
+ except:
+ # detection problem
+ return (False, False, False)
+
+ # Flash fallback
+ if ua.find("MSIE 8") >= 0 or ua.find("MSIE 9") >= 0:
+ return (True, True, True)
+
+ # unsupported
+ return (False, False, True)
+
+
+ ## iOS
+ ##
+ if ua.find("iPhone") >= 0 or ua.find("iPad") >= 0 or ua.find("iPod") >= 0:
+ ## native Hixie76 (as of March 2012), no Flash, no alternative browsers
+ return (True, False, True)
+
+
+ ## Android
+ ##
+ if ua.find("Android") >= 0:
+
+ ## Firefox Mobile
+ ##
+ if ua.find("Firefox") >= 0:
+ # Hybi-10+ for FF Mobile 8+
+ return (True, False, True)
+
+ ## Opera Mobile
+ ##
+ if ua.find("Opera") >= 0:
+ # Hixie76 for Opera 11+
+ return (True, False, True)
+
+ ## Chrome for Android
+ ##
+ if ua.find("CrMo") >= 0:
+ # http://code.google.com/chrome/mobile/docs/faq.html
+ return (True, False, True)
+
+ ## Android builtin Browser (ooold WebKit)
+ ##
+ if ua.find("AppleWebKit") >= 0:
+
+ # Though we return WS = True, and Flash = True here, when the device has no actual Flash support, that
+ # will get later detected in JS. This applies to i.e. ARMv6 devices like Samsung Galaxy ACE
+
+ # builtin browser, only works via Flash
+ return (True, True, True)
+
+ # detection problem
+ return (False, False, False)
+
+
+ ## webOS
+ ##
+ if ua.find("hpwOS") >= 0 or ua.find("webos") >= 0:
+ try:
+ if ua.find("hpwOS") >= 0:
+ vv = [int(x) for x in UA_HPWEBOS.match(ua).groups()[0].split('.')]
+ if vv[0] >= 3:
+ return (True, False, True)
+ elif ua.find("webos") >= 0:
+ vv = [int(x) for x in UA_WEBOS.match(ua).groups()[0].split('.')]
+ if vv[0] >= 2:
+ return (True, False, True)
+ except:
+ # detection problem
+ return (False, False, False)
+ else:
+ # unsupported
+ return (False, False, True)
+
+
+ ## Opera
+ ##
+ if ua.find("Opera") >= 0:
+ # Opera 11+ has Hixie76 (needs to be manually activated though)
+ return (True, False, True)
+
+
+ ## Firefox
+ ##
+ if ua.find("Firefox") >= 0:
+ r = UA_FIREFOX.match(ua)
+ try:
+ v = int(r.groups()[0])
+ if v >= 7:
+ # native Hybi-10+
+ return (True, False, True)
+ elif v >= 3:
+ # works with Flash bridge
+ return (True, True, True)
+ else:
+ # unsupported
+ return (False, False, True)
+ except:
+ # detection problem
+ return (False, False, False)
+
+
+ ## Safari
+ ##
+ if ua.find("Safari") >= 0 and not ua.find("Chrome") >= 0:
+
+ # rely on at least Hixie76
+ return (True, False, True)
+
+
+ ## Chrome
+ ##
+ if ua.find("Chrome") >= 0:
+ r = UA_CHROME.match(ua)
+ try:
+ v = int(r.groups()[0])
+ if v >= 14:
+ # native Hybi-10+
+ return (True, False, True)
+ elif v >= 4:
+ # works with Flash bridge
+ return (True, True, True)
+ else:
+ # unsupported
+ return (False, False, True)
+ except:
+ # detection problem
+ return (False, False, False)
+
+
+ # detection problem
+ return (False, False, False)
+
+
+UA_DETECT_WS_SUPPORT_DB = {}
+
+def lookupWsSupport(ua, debug = True):
+ ws = _lookupWsSupport(ua)
+ if debug:
+ if not UA_DETECT_WS_SUPPORT_DB.has_key(ua):
+ UA_DETECT_WS_SUPPORT_DB[ua] = ws
+
+ if not ws[2]:
+ msg = "UNDETECTED"
+ elif ws[0]:
+ msg = "SUPPORTED"
+ elif not ws[0]:
+ msg = "UNSUPPORTED"
+ else:
+ msg = "ERROR"
+
+ print "DETECT_WS_SUPPORT", ua, ws[0], ws[1], ws[2], msg
+
+ return ws
diff --git a/autobahn/utf8validator.py b/autobahn/utf8validator.py new file mode 100644 index 0000000..2b9b3ec --- /dev/null +++ b/autobahn/utf8validator.py @@ -0,0 +1,111 @@ +###############################################################################
+##
+## Copyright 2011 Tavendo GmbH
+##
+## Note:
+##
+## This code is a Python implementation of the algorithm
+##
+## "Flexible and Economical UTF-8 Decoder"
+##
+## by Bjoern Hoehrmann
+##
+## bjoern@hoehrmann.de
+## http://bjoern.hoehrmann.de/utf-8/decoder/dfa/
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+
+class Utf8Validator:
+ """
+ Incremental UTF-8 validator with constant memory consumption (minimal state).
+
+ Implements the algorithm "Flexible and Economical UTF-8 Decoder" by
+ Bjoern Hoehrmann (http://bjoern.hoehrmann.de/utf-8/decoder/dfa/).
+ """
+
+ ## DFA transitions
+ UTF8VALIDATOR_DFA = [
+ 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, # 00..1f
+ 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, # 20..3f
+ 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, # 40..5f
+ 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, # 60..7f
+ 1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,9,9,9,9,9,9,9,9,9,9,9,9,9,9,9,9, # 80..9f
+ 7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7, # a0..bf
+ 8,8,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2, # c0..df
+ 0xa,0x3,0x3,0x3,0x3,0x3,0x3,0x3,0x3,0x3,0x3,0x3,0x3,0x4,0x3,0x3, # e0..ef
+ 0xb,0x6,0x6,0x6,0x5,0x8,0x8,0x8,0x8,0x8,0x8,0x8,0x8,0x8,0x8,0x8, # f0..ff
+ 0x0,0x1,0x2,0x3,0x5,0x8,0x7,0x1,0x1,0x1,0x4,0x6,0x1,0x1,0x1,0x1, # s0..s0
+ 1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,0,1,1,1,1,1,0,1,0,1,1,1,1,1,1, # s1..s2
+ 1,2,1,1,1,1,1,2,1,2,1,1,1,1,1,1,1,1,1,1,1,1,1,2,1,1,1,1,1,1,1,1, # s3..s4
+ 1,2,1,1,1,1,1,1,1,2,1,1,1,1,1,1,1,1,1,1,1,1,1,3,1,3,1,1,1,1,1,1, # s5..s6
+ 1,3,1,1,1,1,1,3,1,3,1,1,1,1,1,1,1,3,1,1,1,1,1,1,1,1,1,1,1,1,1,1, # s7..s8
+ ]
+
+ UTF8_ACCEPT = 0
+ UTF8_REJECT = 1
+
+ def __init__(self):
+ self.reset()
+
+ def decode(self, b):
+ """
+ Eat one UTF-8 octet, and validate on the fly.
+
+ Returns UTF8_ACCEPT when enough octets have been consumed, in which case
+ self.codepoint contains the decoded Unicode code point.
+
+ Returns UTF8_REJECT when invalid UTF-8 was encountered.
+
+ Returns some other positive integer when more octets need to be eaten.
+ """
+ type = Utf8Validator.UTF8VALIDATOR_DFA[b]
+ if self.state != Utf8Validator.UTF8_ACCEPT:
+ self.codepoint = (b & 0x3f) | (self.codepoint << 6)
+ else:
+ self.codepoint = (0xff >> type) & b
+ self.state = Utf8Validator.UTF8VALIDATOR_DFA[256 + self.state * 16 + type]
+ return self.state
+
+ def reset(self):
+ """
+ Reset validator to start new incremental UTF-8 decode/validation.
+ """
+ self.state = Utf8Validator.UTF8_ACCEPT
+ self.codepoint = 0
+ self.i = 0
+
+ def validate(self, ba):
+ """
+ Incrementally validate a chunk of bytes provided as string.
+
+ Will return a quad (valid?, endsOnCodePoint?, currentIndex, totalIndex).
+
+ As soon as an octet is encountered which renders the octet sequence
+ invalid, a quad with valid? == False is returned. currentIndex returns
+ the index within the currently consumed chunk, and totalIndex the
+ index within the total consumed sequence that was the point of bail out.
+ When valid? == True, currentIndex will be len(ba) and totalIndex the
+ total amount of consumed bytes.
+ """
+ l = len(ba)
+ for i in xrange(l):
+ ## optimized version of decode(), since we are not interested in actual code points
+ self.state = Utf8Validator.UTF8VALIDATOR_DFA[256 + (self.state << 4) + Utf8Validator.UTF8VALIDATOR_DFA[ord(ba[i])]]
+ if self.state == Utf8Validator.UTF8_REJECT:
+ self.i += i
+ return False, False, i, self.i
+ self.i += l
+ return True, self.state == Utf8Validator.UTF8_ACCEPT, l, self.i
diff --git a/autobahn/util.py b/autobahn/util.py new file mode 100644 index 0000000..208ee42 --- /dev/null +++ b/autobahn/util.py @@ -0,0 +1,60 @@ +###############################################################################
+##
+## Copyright 2011 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+import datetime
+import time
+import random
+
+UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+
+
+def utcnow():
+ """
+ Get current time in UTC as ISO 8601 string.
+ """
+ now = datetime.datetime.utcnow()
+ return now.strftime(UTC_TIMESTAMP_FORMAT)
+
+
+def parseutc(s):
+ """
+ Parse an ISO 8601 combined date and time string, like i.e. 2011-11-23T12:23Z
+ into a UTC datetime instance.
+ """
+ try:
+ return datetime.datetime.strptime(s, UTC_TIMESTAMP_FORMAT)
+ except:
+ return None
+
+
+def utcstr(dt):
+ """
+ Convert an UTC datetime instance into an ISO 8601 combined date and time,
+ like i.e. 2011-11-23T12:23Z
+ """
+ try:
+ return dt.strftime(UTC_TIMESTAMP_FORMAT)
+ except:
+ return None
+
+
+def newid():
+ """
+ Generate a new random object ID.
+ """
+ return ''.join([random.choice("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_") for i in xrange(16)])
diff --git a/autobahn/wamp.py b/autobahn/wamp.py new file mode 100644 index 0000000..ca00223 --- /dev/null +++ b/autobahn/wamp.py @@ -0,0 +1,1719 @@ +###############################################################################
+##
+## Copyright 2011,2012 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+import json
+import random
+import inspect, types
+import traceback
+
+import hashlib, hmac, binascii
+
+from twisted.python import log
+from twisted.internet import reactor
+from twisted.internet.defer import Deferred, maybeDeferred
+
+import autobahn
+
+from websocket import WebSocketProtocol, HttpException
+from websocket import WebSocketClientProtocol, WebSocketClientFactory
+from websocket import WebSocketServerFactory, WebSocketServerProtocol
+
+from httpstatus import HTTP_STATUS_CODE_BAD_REQUEST
+from prefixmap import PrefixMap
+from util import utcstr, utcnow, parseutc, newid
+
+
+def exportRpc(arg = None):
+ """
+ Decorator for RPC'ed callables.
+ """
+ ## decorator without argument
+ if type(arg) is types.FunctionType:
+ arg._autobahn_rpc_id = arg.__name__
+ return arg
+ ## decorator with argument
+ else:
+ def inner(f):
+ f._autobahn_rpc_id = arg
+ return f
+ return inner
+
+def exportSub(arg, prefixMatch = False):
+ """
+ Decorator for subscription handlers.
+ """
+ def inner(f):
+ f._autobahn_sub_id = arg
+ f._autobahn_sub_prefix_match = prefixMatch
+ return f
+ return inner
+
+def exportPub(arg, prefixMatch = False):
+ """
+ Decorator for publication handlers.
+ """
+ def inner(f):
+ f._autobahn_pub_id = arg
+ f._autobahn_pub_prefix_match = prefixMatch
+ return f
+ return inner
+
+
+class WampProtocol:
+ """
+ WAMP protocol base class. Mixin for WampServerProtocol and WampClientProtocol.
+ """
+
+ WAMP_PROTOCOL_VERSION = 1
+ """
+ WAMP version this server speaks. Versions are numbered consecutively
+ (integers, no gaps).
+ """
+
+ MESSAGE_TYPEID_WELCOME = 0
+ """
+ Server-to-client welcome message containing session ID.
+ """
+
+ MESSAGE_TYPEID_PREFIX = 1
+ """
+ Client-to-server message establishing a URI prefix to be used in CURIEs.
+ """
+
+ MESSAGE_TYPEID_CALL = 2
+ """
+ Client-to-server message initiating an RPC.
+ """
+
+ MESSAGE_TYPEID_CALL_RESULT = 3
+ """
+ Server-to-client message returning the result of a successful RPC.
+ """
+
+ MESSAGE_TYPEID_CALL_ERROR = 4
+ """
+ Server-to-client message returning the error of a failed RPC.
+ """
+
+ MESSAGE_TYPEID_SUBSCRIBE = 5
+ """
+ Client-to-server message subscribing to a topic.
+ """
+
+ MESSAGE_TYPEID_UNSUBSCRIBE = 6
+ """
+ Client-to-server message unsubscribing from a topic.
+ """
+
+ MESSAGE_TYPEID_PUBLISH = 7
+ """
+ Client-to-server message publishing an event to a topic.
+ """
+
+ MESSAGE_TYPEID_EVENT = 8
+ """
+ Server-to-client message providing the event of a (subscribed) topic.
+ """
+
+
+ ERROR_URI_BASE = "http://autobahn.tavendo.de/error#"
+
+ ERROR_URI_GENERIC = ERROR_URI_BASE + "generic"
+ ERROR_DESC_GENERIC = "generic error"
+
+ ERROR_URI_INTERNAL = ERROR_URI_BASE + "internal"
+ ERROR_DESC_INTERNAL = "internal error"
+
+
+ def connectionMade(self):
+ self.debugWamp = self.factory.debugWamp
+ self.debugApp = self.factory.debugApp
+ self.prefixes = PrefixMap()
+
+
+ def connectionLost(self, reason):
+ pass
+
+
+ def _protocolError(self, reason):
+ if self.debugWamp:
+ log.msg("Closing Wamp session on protocol violation : %s" % reason)
+
+ ## FIXME: subprotocols are probably not supposed to close with CLOSE_STATUS_CODE_PROTOCOL_ERROR
+ ##
+ self.protocolViolation("Wamp RPC/PubSub protocol violation ('%s')" % reason)
+
+
+ def shrink(self, uri, passthrough = False):
+ """
+ Shrink given URI to CURIE according to current prefix mapping.
+ If no appropriate prefix mapping is available, return original URI.
+
+ :param uri: URI to shrink.
+ :type uri: str
+
+ :returns str -- CURIE or original URI.
+ """
+ return self.prefixes.shrink(uri)
+
+
+ def resolve(self, curieOrUri, passthrough = False):
+ """
+ Resolve given CURIE/URI according to current prefix mapping or return
+ None if cannot be resolved.
+
+ :param curieOrUri: CURIE or URI.
+ :type curieOrUri: str
+
+ :returns: str -- Full URI for CURIE or None.
+ """
+ return self.prefixes.resolve(curieOrUri)
+
+
+ def resolveOrPass(self, curieOrUri):
+ """
+ Resolve given CURIE/URI according to current prefix mapping or return
+ string verbatim if cannot be resolved.
+
+ :param curieOrUri: CURIE or URI.
+ :type curieOrUri: str
+
+ :returns: str -- Full URI for CURIE or original string.
+ """
+ return self.prefixes.resolveOrPass(curieOrUri)
+
+
+
+class WampFactory:
+ """
+ WAMP factory base class. Mixin for WampServerFactory and WampClientFactory.
+ """
+
+ pass
+
+
+
+class WampServerProtocol(WebSocketServerProtocol, WampProtocol):
+ """
+ Server factory for Wamp RPC/PubSub.
+ """
+
+ SUBSCRIBE = 1
+ PUBLISH = 2
+
+ def onSessionOpen(self):
+ """
+ Callback fired when WAMP session was fully established.
+ """
+ pass
+
+
+ def onOpen(self):
+ """
+ Default implementation for WAMP connection opened sends
+ Welcome message containing session ID.
+ """
+ self.session_id = newid()
+ msg = [WampProtocol.MESSAGE_TYPEID_WELCOME,
+ self.session_id,
+ WampProtocol.WAMP_PROTOCOL_VERSION,
+ "Autobahn/%s" % autobahn.version]
+ o = json.dumps(msg)
+ self.sendMessage(o)
+ self.factory._addSession(self, self.session_id)
+ self.onSessionOpen()
+
+
+ def onConnect(self, connectionRequest):
+ """
+ Default implementation for WAMP connection acceptance:
+ check if client announced WAMP subprotocol, and only accept connection
+ if client did so.
+ """
+ for p in connectionRequest.protocols:
+ if p in self.factory.protocols:
+ return p
+ raise HttpException(HTTP_STATUS_CODE_BAD_REQUEST[0], "this server only speaks WAMP")
+
+
+ def connectionMade(self):
+ WebSocketServerProtocol.connectionMade(self)
+ WampProtocol.connectionMade(self)
+
+ ## RPCs registered in this session (a URI map of (object, procedure)
+ ## pairs for object methods or (None, procedure) for free standing procedures)
+ self.procs = {}
+
+ ## Publication handlers registered in this session (a URI map of (object, pubHandler) pairs
+ ## pairs for object methods (handlers) or (None, None) for topic without handler)
+ self.pubHandlers = {}
+
+ ## Subscription handlers registered in this session (a URI map of (object, subHandler) pairs
+ ## pairs for object methods (handlers) or (None, None) for topic without handler)
+ self.subHandlers = {}
+
+
+ def connectionLost(self, reason):
+ self.factory._unsubscribeClient(self)
+ self.factory._removeSession(self)
+
+ WampProtocol.connectionLost(self, reason)
+ WebSocketServerProtocol.connectionLost(self, reason)
+
+
+ def sendMessage(self, payload):
+ if self.debugWamp:
+ log.msg("TX WAMP: %s" % str(payload))
+ WebSocketServerProtocol.sendMessage(self, payload)
+
+
+ def _getPubHandler(self, topicUri):
+ ## Longest matching prefix based resolution of (full) topic URI to
+ ## publication handler.
+ ## Returns a 5-tuple (consumedUriPart, unconsumedUriPart, handlerObj, handlerProc, prefixMatch)
+ ##
+ for i in xrange(len(topicUri), -1, -1):
+ tt = topicUri[:i]
+ if self.pubHandlers.has_key(tt):
+ h = self.pubHandlers[tt]
+ return (tt, topicUri[i:], h[0], h[1], h[2])
+ return None
+
+
+ def _getSubHandler(self, topicUri):
+ ## Longest matching prefix based resolution of (full) topic URI to
+ ## subscription handler.
+ ## Returns a 5-tuple (consumedUriPart, unconsumedUriPart, handlerObj, handlerProc, prefixMatch)
+ ##
+ for i in xrange(len(topicUri), -1, -1):
+ tt = topicUri[:i]
+ if self.subHandlers.has_key(tt):
+ h = self.subHandlers[tt]
+ return (tt, topicUri[i:], h[0], h[1], h[2])
+ return None
+
+
+ def registerForPubSub(self, topicUri, prefixMatch = False, pubsub = PUBLISH | SUBSCRIBE):
+ """
+ Register a topic URI as publish/subscribe channel in this session.
+
+ :param topicUri: Topic URI to be established as publish/subscribe channel.
+ :type topicUri: str
+ :param prefixMatch: Allow to match this topic URI by prefix.
+ :type prefixMatch: bool
+ :param pubsub: Allow publication and/or subscription.
+ :type pubsub: WampServerProtocol.PUB, WampServerProtocol.SUB, WampServerProtocol.PUB | WampServerProtocol.SUB
+ """
+ if pubsub & WampServerProtocol.PUBLISH:
+ self.pubHandlers[topicUri] = (None, None, prefixMatch)
+ if self.debugWamp:
+ log.msg("registered topic %s for publication (match by prefix = %s)" % (topicUri, prefixMatch))
+ if pubsub & WampServerProtocol.SUBSCRIBE:
+ self.subHandlers[topicUri] = (None, None, prefixMatch)
+ if self.debugWamp:
+ log.msg("registered topic %s for subscription (match by prefix = %s)" % (topicUri, prefixMatch))
+
+
+ def registerHandlerForPubSub(self, obj, baseUri = ""):
+ """
+ Register a handler object for PubSub. A handler object has methods
+ which are decorated using @exportPub and @exportSub.
+
+ :param obj: The object to be registered (in this WebSockets session) for PubSub.
+ :type obj: Object with methods decorated using @exportPub and @exportSub.
+ :param baseUri: Optional base URI which is prepended to topic names for export.
+ :type baseUri: String.
+ """
+ for k in inspect.getmembers(obj.__class__, inspect.ismethod):
+ if k[1].__dict__.has_key("_autobahn_pub_id"):
+ uri = baseUri + k[1].__dict__["_autobahn_pub_id"]
+ prefixMatch = k[1].__dict__["_autobahn_pub_prefix_match"]
+ proc = k[1]
+ self.registerHandlerForPub(uri, obj, proc, prefixMatch)
+ elif k[1].__dict__.has_key("_autobahn_sub_id"):
+ uri = baseUri + k[1].__dict__["_autobahn_sub_id"]
+ prefixMatch = k[1].__dict__["_autobahn_sub_prefix_match"]
+ proc = k[1]
+ self.registerHandlerForSub(uri, obj, proc, prefixMatch)
+
+
+ def registerHandlerForSub(self, uri, obj, proc, prefixMatch = False):
+ """
+ Register a method of an object as subscription handler.
+
+ :param uri: Topic URI to register subscription handler for.
+ :type uri: str
+ :param obj: The object on which to register a method as subscription handler.
+ :type obj: object
+ :param proc: Unbound object method to register as subscription handler.
+ :type proc: unbound method
+ :param prefixMatch: Allow to match this topic URI by prefix.
+ :type prefixMatch: bool
+ """
+ self.subHandlers[uri] = (obj, proc, prefixMatch)
+ if not self.pubHandlers.has_key(uri):
+ self.pubHandlers[uri] = (None, None, False)
+ if self.debugWamp:
+ log.msg("registered subscription handler for topic %s" % uri)
+
+
+ def registerHandlerForPub(self, uri, obj, proc, prefixMatch = False):
+ """
+ Register a method of an object as publication handler.
+
+ :param uri: Topic URI to register publication handler for.
+ :type uri: str
+ :param obj: The object on which to register a method as publication handler.
+ :type obj: object
+ :param proc: Unbound object method to register as publication handler.
+ :type proc: unbound method
+ :param prefixMatch: Allow to match this topic URI by prefix.
+ :type prefixMatch: bool
+ """
+ self.pubHandlers[uri] = (obj, proc, prefixMatch)
+ if not self.subHandlers.has_key(uri):
+ self.subHandlers[uri] = (None, None, False)
+ if self.debugWamp:
+ log.msg("registered publication handler for topic %s" % uri)
+
+
+ def registerForRpc(self, obj, baseUri = "", methods = None):
+ """
+ Register an service object for RPC. A service object has methods
+ which are decorated using @exportRpc.
+
+ :param obj: The object to be registered (in this WebSockets session) for RPC.
+ :type obj: Object with methods decorated using @exportRpc.
+ :param baseUri: Optional base URI which is prepended to method names for export.
+ :type baseUri: String.
+ :param methods: If not None, a list of unbound class methods corresponding to obj
+ which should be registered. This can be used to register only a subset
+ of the methods decorated with @exportRpc.
+ :type methods: List of unbound class methods.
+ """
+ for k in inspect.getmembers(obj.__class__, inspect.ismethod):
+ if k[1].__dict__.has_key("_autobahn_rpc_id"):
+ if methods is None or k[1] in methods:
+ uri = baseUri + k[1].__dict__["_autobahn_rpc_id"]
+ proc = k[1]
+ self.registerMethodForRpc(uri, obj, proc)
+
+
+ def registerMethodForRpc(self, uri, obj, proc):
+ """
+ Register a method of an object for RPC.
+
+ :param uri: URI to register RPC method under.
+ :type uri: str
+ :param obj: The object on which to register a method for RPC.
+ :type obj: object
+ :param proc: Unbound object method to register RPC for.
+ :type proc: unbound method
+ """
+ self.procs[uri] = (obj, proc)
+ if self.debugWamp:
+ log.msg("registered remote procedure on %s" % uri)
+
+
+ def registerProcedureForRpc(self, uri, proc):
+ """
+ Register a (free standing) function/procedure for RPC.
+
+ :param uri: URI to register RPC function/procedure under.
+ :type uri: str
+ :param proc: Free-standing function/procedure.
+ :type proc: function/procedure
+ """
+ self.procs[uri] = (None, proc)
+ if self.debugWamp:
+ log.msg("registered remote procedure on %s" % uri)
+
+
+ def dispatch(self, topicUri, event, exclude = [], eligible = None):
+ """
+ Dispatch an event for a topic to all clients subscribed to
+ and authorized for that topic.
+
+ Optionally, exclude list of clients and/or only consider clients
+ from explicit eligibles. In other words, the event is delivered
+ to the set
+
+ (subscribers - excluded) & eligible
+
+ :param topicUri: URI of topic to publish event to.
+ :type topicUri: str
+ :param event: Event to dispatch.
+ :type event: obj
+ :param exclude: Optional list of clients (WampServerProtocol instances) to exclude.
+ :type exclude: list of obj
+ :param eligible: Optional list of clients (WampServerProtocol instances) eligible at all (or None for all).
+ :type eligible: list of obj
+ """
+ self.factory.dispatch(topicUri, event, exclude, eligible)
+
+
+ def _callProcedure(self, uri, arg = None):
+ """
+ INTERNAL METHOD! Actually performs the call of a procedure invoked via RPC.
+ """
+ if self.procs.has_key(uri):
+ m = self.procs[uri]
+ if arg:
+ ## method/function called with args
+ args = tuple(arg)
+ if m[0]:
+ ## call object method
+ return m[1](m[0], *args)
+ else:
+ ## call free-standing function/procedure
+ return m[1](*args)
+ else:
+ ## method/function called without args
+ if m[0]:
+ ## call object method
+ return m[1](m[0])
+ else:
+ ## call free-standing function/procedure
+ return m[1]()
+ else:
+ raise Exception("no procedure %s" % uri)
+
+
+ def _sendCallResult(self, result, callid):
+ """
+ INTERNAL METHOD! Marshal and send a RPC success result.
+ """
+ msg = [WampProtocol.MESSAGE_TYPEID_CALL_RESULT, callid, result]
+ try:
+ rmsg = json.dumps(msg)
+ except:
+ raise Exception("call result not JSON serializable")
+ else:
+ self.sendMessage(rmsg)
+
+
+ def _sendCallError(self, error, callid):
+ """
+ INTERNAL METHOD! Marshal and send a RPC error result.
+ """
+ try:
+
+ eargs = error.value.args
+ leargs = len(eargs)
+ traceb = error.getTraceback()
+
+ if leargs == 0:
+ erroruri = WampProtocol.ERROR_URI_GENERIC
+ errordesc = WampProtocol.ERROR_DESC_GENERIC
+ errordetails = None
+
+ elif leargs == 1:
+ if type(eargs[0]) not in [str, unicode]:
+ raise Exception("invalid type %s for errorDesc" % type(eargs[0]))
+ erroruri = WampProtocol.ERROR_URI_GENERIC
+ errordesc = eargs[0]
+ errordetails = None
+
+ elif leargs in [2, 3]:
+ if type(eargs[0]) not in [str, unicode]:
+ raise Exception("invalid type %s for errorUri" % type(eargs[0]))
+ erroruri = eargs[0]
+ if type(eargs[1]) not in [str, unicode]:
+ raise Exception("invalid type %s for errorDesc" % type(eargs[1]))
+ errordesc = eargs[1]
+ if leargs > 2:
+ errordetails = eargs[2] # this must be JSON serializable .. if not, we get exception later in sendMessage
+ else:
+ errordetails = None
+
+ else:
+ raise Exception("invalid args length %d for exception" % leargs)
+
+ if errordetails is not None:
+ msg = [WampProtocol.MESSAGE_TYPEID_CALL_ERROR, callid, self.prefixes.shrink(erroruri), errordesc, errordetails]
+ else:
+ msg = [WampProtocol.MESSAGE_TYPEID_CALL_ERROR, callid, self.prefixes.shrink(erroruri), errordesc]
+
+ try:
+ rmsg = json.dumps(msg)
+ except Exception, e:
+ raise Exception("invalid object for errorDetails - not JSON serializable (%s)" % str(e))
+
+ if self.debugApp:
+ log.msg("application error")
+ log.msg(traceb)
+ log.msg(msg)
+
+ except Exception, e:
+
+ if self.debugWamp:
+ log.err(str(e))
+ log.err(error.getTraceback())
+
+ msg = [WampProtocol.MESSAGE_TYPEID_CALL_ERROR, callid, self.prefixes.shrink(WampProtocol.ERROR_URI_INTERNAL), WampProtocol.ERROR_DESC_INTERNAL]
+ rmsg = json.dumps(msg)
+
+ finally:
+
+ self.sendMessage(rmsg)
+
+
+ def onMessage(self, msg, binary):
+ """
+ INTERNAL METHOD! Handle WAMP messages received from WAMP client.
+ """
+
+ if self.debugWamp:
+ log.msg("RX WAMP: %s" % str(msg))
+
+ if not binary:
+ try:
+ obj = json.loads(msg)
+ if type(obj) == list:
+
+ ## Call Message
+ ##
+ if obj[0] == WampProtocol.MESSAGE_TYPEID_CALL:
+ callid = obj[1]
+ procuri = self.prefixes.resolveOrPass(obj[2])
+ arg = obj[3:]
+ d = maybeDeferred(self._callProcedure, procuri, arg)
+ d.addCallback(self._sendCallResult, callid)
+ d.addErrback(self._sendCallError, callid)
+
+ ## Subscribe Message
+ ##
+ elif obj[0] == WampProtocol.MESSAGE_TYPEID_SUBSCRIBE:
+ topicUri = self.prefixes.resolveOrPass(obj[1])
+ h = self._getSubHandler(topicUri)
+ if h:
+ ## either exact match or prefix match allowed
+ if h[1] == "" or h[4]:
+
+ ## direct topic
+ if h[2] is None and h[3] is None:
+ self.factory._subscribeClient(self, topicUri)
+
+ ## topic handled by subscription handler
+ else:
+ try:
+ ## handler is object method
+ if h[2]:
+ a = h[3](h[2], str(h[0]), str(h[1]))
+
+ ## handler is free standing procedure
+ else:
+ a = h[3](str(h[0]), str(h[1]))
+
+ ## only subscribe client if handler did return True
+ if a:
+ self.factory._subscribeClient(self, topicUri)
+ except:
+ if self.debugWamp:
+ log.msg("execption during topic subscription handler")
+ else:
+ if self.debugWamp:
+ log.msg("topic %s matches only by prefix and prefix match disallowed" % topicUri)
+ else:
+ if self.debugWamp:
+ log.msg("no topic / subscription handler registered for %s" % topicUri)
+
+ ## Unsubscribe Message
+ ##
+ elif obj[0] == WampProtocol.MESSAGE_TYPEID_UNSUBSCRIBE:
+ topicUri = self.prefixes.resolveOrPass(obj[1])
+ self.factory._unsubscribeClient(self, topicUri)
+
+ ## Publish Message
+ ##
+ elif obj[0] == WampProtocol.MESSAGE_TYPEID_PUBLISH:
+ topicUri = self.prefixes.resolveOrPass(obj[1])
+ h = self._getPubHandler(topicUri)
+ if h:
+ ## either exact match or prefix match allowed
+ if h[1] == "" or h[4]:
+
+ ## Event
+ ##
+ event = obj[2]
+
+ ## Exclude Sessions List
+ ##
+ exclude = [self] # exclude publisher by default
+ if len(obj) >= 4:
+ if type(obj[3]) == bool:
+ if not obj[3]:
+ exclude = []
+ elif type(obj[3]) == list:
+ ## map session IDs to protos
+ exclude = self.factory.sessionIdsToProtos(obj[3])
+ else:
+ ## FIXME: invalid type
+ pass
+
+ ## Eligible Sessions List
+ ##
+ eligible = None # all sessions are eligible by default
+ if len(obj) >= 5:
+ if type(obj[4]) == list:
+ ## map session IDs to protos
+ eligible = self.factory.sessionIdsToProtos(obj[4])
+ else:
+ ## FIXME: invalid type
+ pass
+
+ ## direct topic
+ if h[2] is None and h[3] is None:
+ self.factory.dispatch(topicUri, event, exclude, eligible)
+
+ ## topic handled by publication handler
+ else:
+ try:
+ ## handler is object method
+ if h[2]:
+ e = h[3](h[2], str(h[0]), str(h[1]), event)
+
+ ## handler is free standing procedure
+ else:
+ e = h[3](str(h[0]), str(h[1]), event)
+
+ ## only dispatch event if handler did return event
+ if e:
+ self.factory.dispatch(topicUri, e, exclude, eligible)
+ except:
+ if self.debugWamp:
+ log.msg("execption during topic publication handler")
+ else:
+ if self.debugWamp:
+ log.msg("topic %s matches only by prefix and prefix match disallowed" % topicUri)
+ else:
+ if self.debugWamp:
+ log.msg("no topic / publication handler registered for %s" % topicUri)
+
+ ## Define prefix to be used in CURIEs
+ ##
+ elif obj[0] == WampProtocol.MESSAGE_TYPEID_PREFIX:
+ prefix = obj[1]
+ uri = obj[2]
+ self.prefixes.set(prefix, uri)
+
+ else:
+ log.msg("unknown message type")
+ else:
+ log.msg("msg not a list")
+ except Exception, e:
+ traceback.print_exc()
+ else:
+ log.msg("binary message")
+
+
+
+class WampServerFactory(WebSocketServerFactory, WampFactory):
+ """
+ Server factory for Wamp RPC/PubSub.
+ """
+
+ protocol = WampServerProtocol
+ """
+ Twisted protocol used by default for WAMP servers.
+ """
+
+ def __init__(self, url, debug = False, debugCodePaths = False, debugWamp = False, debugApp = False):
+ WebSocketServerFactory.__init__(self, url, protocols = ["wamp"], debug = debug, debugCodePaths = debugCodePaths)
+ self.debugWamp = debugWamp
+ self.debugApp = debugApp
+
+
+ def _subscribeClient(self, proto, topicUri):
+ """
+ INTERNAL METHOD! Called from proto to subscribe client for topic.
+ """
+
+ if self.debugWamp:
+ log.msg("subscribed peer %s for topic %s" % (proto.peerstr, topicUri))
+
+ if not self.subscriptions.has_key(topicUri):
+ self.subscriptions[topicUri] = set()
+ self.subscriptions[topicUri].add(proto)
+
+
+ def _unsubscribeClient(self, proto, topicUri = None):
+ """
+ INTERNAL METHOD! Called from proto to unsubscribe client from topic.
+ """
+
+ if topicUri:
+ if self.subscriptions.has_key(topicUri):
+ self.subscriptions[topicUri].discard(proto)
+ if self.debugWamp:
+ log.msg("unsubscribed peer %s from topic %s" % (proto.peerstr, topicUri))
+ else:
+ for t in self.subscriptions:
+ self.subscriptions[t].discard(proto)
+ if self.debugWamp:
+ log.msg("unsubscribed peer %s from all topics" % (proto.peerstr))
+
+
+ def dispatch(self, topicUri, event, exclude = [], eligible = None):
+ """
+ Dispatch an event to all peers subscribed to the event topic.
+
+ :param topicUri: Topic to publish event to.
+ :type topicUri: str
+ :param event: Event to publish (must be JSON serializable).
+ :type event: obj
+ :param exclude: List of WampServerProtocol instances to exclude from receivers.
+ :type exclude: List of obj
+ :param eligible: List of WampServerProtocol instances eligible as receivers (or None for all).
+ :type eligible: List of obj
+
+ :returns twisted.internet.defer.Deferred -- Will be fired when event was
+ dispatched to all subscribers. The return value provided to the deferred
+ is a pair (delivered, requested), where delivered = number of actual
+ receivers, and requested = number of (subscribers - excluded) & eligible.
+ """
+ if self.debugWamp:
+ log.msg("publish event %s for topicUri %s" % (str(event), topicUri))
+
+ d = Deferred()
+
+ if self.subscriptions.has_key(topicUri) and len(self.subscriptions[topicUri]) > 0:
+
+ ## FIXME: this might break ordering of event delivery from a
+ ## receiver perspective. We might need to have send queues
+ ## per receiver OR do recvs = deque(sorted(..))
+
+ ## However, see http://twistedmatrix.com/trac/ticket/1396
+
+ if eligible is not None:
+ subscrbs = set(eligible) & self.subscriptions[topicUri]
+ else:
+ subscrbs = self.subscriptions[topicUri]
+
+ if len(exclude) > 0:
+ recvs = subscrbs - set(exclude)
+ else:
+ recvs = subscrbs
+
+ l = len(recvs)
+ if l > 0:
+
+ o = [WampProtocol.MESSAGE_TYPEID_EVENT, topicUri, event]
+ try:
+ msg = json.dumps(o)
+ if self.debugWamp:
+ log.msg("serialized event msg: " + str(msg))
+ except:
+ raise Exception("invalid type for event (not JSON serializable)")
+
+ preparedMsg = self.prepareMessage(msg)
+ self._sendEvents(preparedMsg, recvs.copy(), 0, l, d)
+ else:
+ d.callback((0, 0))
+
+ return d
+
+
+ def _sendEvents(self, preparedMsg, recvs, delivered, requested, d):
+ """
+ INTERNAL METHOD! Delivers events to receivers in chunks and
+ reenters the reactor in-between, so that other stuff can run.
+ """
+ ## deliver a batch of events
+ done = False
+ for i in xrange(0, 256):
+ try:
+ proto = recvs.pop()
+ if proto.state == WebSocketProtocol.STATE_OPEN:
+ try:
+ proto.sendPreparedMessage(preparedMsg)
+ except:
+ pass
+ else:
+ if self.debugWamp:
+ log.msg("delivered event to peer %s" % proto.peerstr)
+ delivered += 1
+ except KeyError:
+ # all receivers done
+ done = True
+ break
+
+ if not done:
+ ## if there are receivers left, redo
+ reactor.callLater(0, self._sendEvents, preparedMsg, recvs, delivered, requested, d)
+ else:
+ ## else fire final result
+ d.callback((delivered, requested))
+
+
+ def _addSession(self, proto, session_id):
+ """
+ INTERNAL METHOD! Add proto for session ID.
+ """
+ if not self.protoToSessions.has_key(proto):
+ self.protoToSessions[proto] = session_id
+ else:
+ raise Exception("logic error - dublicate _addSession for protoToSessions")
+ if not self.sessionsToProto.has_key(session_id):
+ self.sessionsToProto[session_id] = proto
+ else:
+ raise Exception("logic error - dublicate _addSession for sessionsToProto")
+
+
+ def _removeSession(self, proto):
+ """
+ INTERNAL METHOD! Remove session by proto.
+ """
+ if self.protoToSessions.has_key(proto):
+ session_id = self.protoToSessions[proto]
+ del self.protoToSessions[proto]
+ if self.sessionsToProto.has_key(session_id):
+ del self.sessionsToProto[session_id]
+
+
+ def sessionIdsToProtos(self, sessionIds):
+ """
+ Map session IDs to connected client protocol instances.
+
+ :param sessionIds: List of session IDs to be mapped.
+ :type sessionIds: list of str
+
+ :returns list of WampServerProtocol instances -- List of protocol instances corresponding to the session IDs.
+ """
+ protos = []
+ for s in sessionIds:
+ if self.sessionsToProto.has_key(s):
+ protos.append(self.sessionsToProto[s])
+ return protos
+
+
+ def protosToSessionIds(self, protos):
+ """
+ Map connected client protocol instances to session IDs.
+
+ :param protos: List of instances of WampServerProtocol to be mapped.
+ :type protos: list of WampServerProtocol
+
+ :returns list of str -- List of session IDs corresponding to the protos.
+ """
+ sessionIds = []
+ for p in protos:
+ if self.protoToSessions.has_key(p):
+ sessionIds.append(self.protoToSessions[p])
+ return sessionIds
+
+
+ def startFactory(self):
+ """
+ Called by Twisted when the factory starts up. When overriding, make
+ sure to call the base method.
+ """
+ if self.debugWamp:
+ log.msg("WampServerFactory starting")
+ self.subscriptions = {}
+ self.protoToSessions = {}
+ self.sessionsToProto = {}
+
+
+ def stopFactory(self):
+ """
+ Called by Twisted when the factory shuts down. When overriding, make
+ sure to call the base method.
+ """
+ if self.debugWamp:
+ log.msg("WampServerFactory stopped")
+
+
+
+class WampClientProtocol(WebSocketClientProtocol, WampProtocol):
+ """
+ Twisted client protocol for WAMP.
+ """
+
+ def onSessionOpen(self):
+ """
+ Callback fired when WAMP session was fully established. Override
+ in derived class.
+ """
+ pass
+
+
+ def onOpen(self):
+ ## do nothing here .. onSessionOpen is only fired when welcome
+ ## message was received (and thus session ID set)
+ pass
+
+
+ def onConnect(self, connectionResponse):
+ if connectionResponse.protocol not in self.factory.protocols:
+ raise Exception("server does not speak WAMP")
+
+
+ def connectionMade(self):
+ WebSocketClientProtocol.connectionMade(self)
+ WampProtocol.connectionMade(self)
+
+ self.calls = {}
+ self.subscriptions = {}
+
+
+ def connectionLost(self, reason):
+ WampProtocol.connectionLost(self, reason)
+ WebSocketClientProtocol.connectionLost(self, reason)
+
+
+ def sendMessage(self, payload):
+ if self.debugWamp:
+ log.msg("TX WAMP: %s" % str(payload))
+ WebSocketClientProtocol.sendMessage(self, payload)
+
+
+ def onMessage(self, msg, binary):
+ """Internal method to handle WAMP messages received from WAMP server."""
+
+ ## WAMP is text message only
+ ##
+ if binary:
+ self._protocolError("binary WebSocket message received")
+ return
+
+ if self.debugWamp:
+ log.msg("RX WAMP: %s" % str(msg))
+
+ ## WAMP is proper JSON payload
+ ##
+ try:
+ obj = json.loads(msg)
+ except:
+ self._protocolError("WAMP message payload not valid JSON")
+ return
+
+ ## Every WAMP message is a list
+ ##
+ if type(obj) != list:
+ self._protocolError("WAMP message payload not a list")
+ return
+
+ ## Every WAMP message starts with an integer for message type
+ ##
+ if len(obj) < 1:
+ self._protocolError("WAMP message without message type")
+ return
+ if type(obj[0]) != int:
+ self._protocolError("WAMP message type not an integer")
+ return
+
+ ## WAMP message type
+ ##
+ msgtype = obj[0]
+
+ ## Valid WAMP message types received by WAMP clients
+ ##
+ if msgtype not in [WampProtocol.MESSAGE_TYPEID_WELCOME,
+ WampProtocol.MESSAGE_TYPEID_CALL_RESULT,
+ WampProtocol.MESSAGE_TYPEID_CALL_ERROR,
+ WampProtocol.MESSAGE_TYPEID_EVENT]:
+ self._protocolError("invalid WAMP message type %d" % msgtype)
+ return
+
+ ## WAMP CALL_RESULT / CALL_ERROR
+ ##
+ if msgtype in [WampProtocol.MESSAGE_TYPEID_CALL_RESULT, WampProtocol.MESSAGE_TYPEID_CALL_ERROR]:
+
+ ## Call ID
+ ##
+ if len(obj) < 2:
+ self._protocolError("WAMP CALL_RESULT/CALL_ERROR message without <callid>")
+ return
+ if type(obj[1]) not in [unicode, str]:
+ self._protocolError("WAMP CALL_RESULT/CALL_ERROR message with invalid type %s for <callid>" % type(obj[1]))
+ return
+ callid = str(obj[1])
+
+ ## Pop and process Call Deferred
+ ##
+ d = self.calls.pop(callid, None)
+ if d:
+ ## WAMP CALL_RESULT
+ ##
+ if msgtype == WampProtocol.MESSAGE_TYPEID_CALL_RESULT:
+ ## Call Result
+ ##
+ if len(obj) != 3:
+ self._protocolError("WAMP CALL_RESULT message with invalid length %d" % len(obj))
+ return
+ result = obj[2]
+
+ ## Fire Call Success Deferred
+ ##
+ d.callback(result)
+
+ ## WAMP CALL_ERROR
+ ##
+ elif msgtype == WampProtocol.MESSAGE_TYPEID_CALL_ERROR:
+ if len(obj) not in [4, 5]:
+ self._protocolError("call error message invalid length %d" % len(obj))
+ return
+
+ ## Error URI
+ ##
+ if type(obj[2]) not in [unicode, str]:
+ self._protocolError("invalid type %s for errorUri in call error message" % str(type(obj[2])))
+ return
+ erroruri = str(obj[2])
+
+ ## Error Description
+ ##
+ if type(obj[3]) not in [unicode, str]:
+ self._protocolError("invalid type %s for errorDesc in call error message" % str(type(obj[3])))
+ return
+ errordesc = str(obj[3])
+
+ ## Error Details
+ ##
+ if len(obj) > 4:
+ errordetails = obj[4]
+ else:
+ errordetails = None
+
+ ## Fire Call Error Deferred
+ ##
+ e = Exception()
+ e.args = (erroruri, errordesc, errordetails)
+ d.errback(e)
+ else:
+ raise Exception("logic error")
+ else:
+ if self.debugWamp:
+ log.msg("callid not found for received call result/error message")
+
+ ## WAMP EVENT
+ ##
+ elif msgtype == WampProtocol.MESSAGE_TYPEID_EVENT:
+ ## Topic
+ ##
+ if len(obj) != 3:
+ self._protocolError("WAMP EVENT message invalid length %d" % len(obj))
+ return
+ if type(obj[1]) not in [unicode, str]:
+ self._protocolError("invalid type for <topic> in WAMP EVENT message")
+ return
+ unresolvedTopicUri = str(obj[1])
+ topicUri = self.prefixes.resolveOrPass(unresolvedTopicUri)
+
+ ## Fire PubSub Handler
+ ##
+ if self.subscriptions.has_key(topicUri):
+ event = obj[2]
+ self.subscriptions[topicUri](topicUri, event)
+ else:
+ ## event received for non-subscribed topic (could be because we
+ ## just unsubscribed, and server already sent out event for
+ ## previous subscription)
+ pass
+
+ ## WAMP WELCOME
+ ##
+ elif msgtype == WampProtocol.MESSAGE_TYPEID_WELCOME:
+ ## Session ID
+ ##
+ if len(obj) < 2:
+ self._protocolError("WAMP WELCOME message invalid length %d" % len(obj))
+ return
+ if type(obj[1]) not in [unicode, str]:
+ self._protocolError("invalid type for <sessionid> in WAMP WELCOME message")
+ return
+ self.session_id = str(obj[1])
+
+ ## WAMP Protocol Version
+ ##
+ if len(obj) > 2:
+ if type(obj[2]) not in [int]:
+ self._protocolError("invalid type for <version> in WAMP WELCOME message")
+ return
+ else:
+ self.session_protocol_version = obj[2]
+ else:
+ self.session_protocol_version = None
+
+ ## Server Ident
+ ##
+ if len(obj) > 3:
+ if type(obj[3]) not in [unicode, str]:
+ self._protocolError("invalid type for <server> in WAMP WELCOME message")
+ return
+ else:
+ self.session_server = obj[3]
+ else:
+ self.session_server = None
+
+ self.onSessionOpen()
+
+ else:
+ raise Exception("logic error")
+
+
+ def call(self, *args):
+ """
+ Perform a remote-procedure call (RPC). The first argument is the procedure
+ URI (mandatory). Subsequent positional arguments can be provided (must be
+ JSON serializable). The return value is a Twisted Deferred.
+ """
+
+ if len(args) < 1:
+ raise Exception("missing procedure URI")
+
+ if type(args[0]) not in [unicode, str]:
+ raise Exception("invalid type for procedure URI")
+
+ procuri = args[0]
+ while True:
+ callid = newid()
+ if not self.calls.has_key(callid):
+ break
+ d = Deferred()
+ self.calls[callid] = d
+ msg = [WampProtocol.MESSAGE_TYPEID_CALL, callid, procuri]
+ msg.extend(args[1:])
+
+ try:
+ o = json.dumps(msg)
+ except:
+ raise Exception("call argument(s) not JSON serializable")
+
+ self.sendMessage(o)
+ return d
+
+
+ def prefix(self, prefix, uri):
+ """
+ Establishes a prefix to be used in CURIEs instead of URIs having that
+ prefix for both client-to-server and server-to-client messages.
+
+ :param prefix: Prefix to be used in CURIEs.
+ :type prefix: str
+ :param uri: URI that this prefix will resolve to.
+ :type uri: str
+ """
+
+ if type(prefix) != str:
+ raise Exception("invalid type for prefix")
+
+ if type(uri) not in [unicode, str]:
+ raise Exception("invalid type for URI")
+
+ if self.prefixes.get(prefix):
+ raise Exception("prefix already defined")
+
+ self.prefixes.set(prefix, uri)
+
+ msg = [WampProtocol.MESSAGE_TYPEID_PREFIX, prefix, uri]
+
+ self.sendMessage(json.dumps(msg))
+
+
+ def publish(self, topicUri, event, excludeMe = None, exclude = None, eligible = None):
+ """
+ Publish an event under a topic URI. The latter may be abbreviated using a
+ CURIE which has been previously defined using prefix(). The event must
+ be JSON serializable.
+
+ :param topicUri: The topic URI or CURIE.
+ :type topicUri: str
+ :param event: Event to be published (must be JSON serializable) or None.
+ :type event: value
+ :param excludeMe: When True, don't deliver the published event to myself (when I'm subscribed).
+ :type excludeMe: bool
+ :param exclude: Optional list of session IDs to exclude from receivers.
+ :type exclude: list of str
+ :param eligible: Optional list of session IDs to that are eligible as receivers.
+ :type eligible: list of str
+ """
+
+ if type(topicUri) not in [unicode, str]:
+ raise Exception("invalid type for parameter 'topicUri' - must be string (was %s)" % type(topicUri))
+
+ if excludeMe is not None:
+ if type(excludeMe) != bool:
+ raise Exception("invalid type for parameter 'excludeMe' - must be bool (was %s)" % type(excludeMe))
+
+ if exclude is not None:
+ if type(exclude) != list:
+ raise Exception("invalid type for parameter 'exclude' - must be list (was %s)" % type(exclude))
+
+ if eligible is not None:
+ if type(eligible) != list:
+ raise Exception("invalid type for parameter 'eligible' - must be list (was %s)" % type(eligible))
+
+ if exclude is not None or eligible is not None:
+ if exclude is None:
+ if excludeMe is not None:
+ if excludeMe:
+ exclude = [self.session_id]
+ else:
+ exclude = []
+ else:
+ exclude = [self.session_id]
+ if eligible is not None:
+ msg = [WampProtocol.MESSAGE_TYPEID_PUBLISH, topicUri, event, exclude, eligible]
+ else:
+ msg = [WampProtocol.MESSAGE_TYPEID_PUBLISH, topicUri, event, exclude]
+ else:
+ if excludeMe:
+ msg = [WampProtocol.MESSAGE_TYPEID_PUBLISH, topicUri, event]
+ else:
+ msg = [WampProtocol.MESSAGE_TYPEID_PUBLISH, topicUri, event, excludeMe]
+
+ try:
+ o = json.dumps(msg)
+ except:
+ raise Exception("invalid type for parameter 'event' - not JSON serializable")
+
+ self.sendMessage(o)
+
+
+ def subscribe(self, topicUri, handler):
+ """
+ Subscribe to topic. When already subscribed, will overwrite the handler.
+
+ :param topicUri: URI or CURIE of topic to subscribe to.
+ :type topicUri: str
+ :param handler: Event handler to be invoked upon receiving events for topic.
+ :type handler: Python callable, will be called as in <callable>(eventUri, event).
+ """
+ if type(topicUri) not in [unicode, str]:
+ raise Exception("invalid type for parameter 'topicUri' - must be string (was %s)" % type(topicUri))
+
+ if type(handler) not in [types.FunctionType, types.MethodType, types.BuiltinFunctionType, types.BuiltinMethodType]:
+ raise Exception("invalid type for parameter 'handler' - must be a callable (was %s)" % type(handler))
+
+ turi = self.prefixes.resolveOrPass(topicUri)
+ if not self.subscriptions.has_key(turi):
+ msg = [WampProtocol.MESSAGE_TYPEID_SUBSCRIBE, topicUri]
+ o = json.dumps(msg)
+ self.sendMessage(o)
+ self.subscriptions[turi] = handler
+
+
+ def unsubscribe(self, topicUri):
+ """
+ Unsubscribe from topic. Will do nothing when currently not subscribed to the topic.
+
+ :param topicUri: URI or CURIE of topic to unsubscribe from.
+ :type topicUri: str
+ """
+ if type(topicUri) not in [unicode, str]:
+ raise Exception("invalid type for parameter 'topicUri' - must be string (was %s)" % type(topicUri))
+
+ turi = self.prefixes.resolveOrPass(topicUri)
+ if self.subscriptions.has_key(turi):
+ msg = [WampProtocol.MESSAGE_TYPEID_UNSUBSCRIBE, topicUri]
+ o = json.dumps(msg)
+ self.sendMessage(o)
+ del self.subscriptions[turi]
+
+
+
+class WampClientFactory(WebSocketClientFactory, WampFactory):
+ """
+ Twisted client factory for WAMP.
+ """
+
+ protocol = WampClientProtocol
+
+ def __init__(self, url, debug = False, debugCodePaths = False, debugWamp = False, debugApp = False):
+ WebSocketClientFactory.__init__(self, url, protocols = ["wamp"], debug = debug, debugCodePaths = debugCodePaths)
+ self.debugWamp = debugWamp
+ self.debugApp = debugApp
+
+
+ def startFactory(self):
+ """
+ Called by Twisted when the factory starts up. When overriding, make
+ sure to call the base method.
+ """
+ if self.debugWamp:
+ log.msg("WebSocketClientFactory starting")
+
+
+ def stopFactory(self):
+ """
+ Called by Twisted when the factory shuts down. When overriding, make
+ sure to call the base method.
+ """
+ if self.debugWamp:
+ log.msg("WebSocketClientFactory stopped")
+
+
+
+class WampCraProtocol:
+ """
+ Base class for WAMP Challenge-Response Authentication protocols (client and server).
+
+ WAMP-CRA is a cryptographically strong challenge response authentication
+ protocol based on HMAC-SHA256.
+
+ The protocol performs in-band authentication of WAMP clients to WAMP servers.
+
+ WAMP-CRA does not introduce any new WAMP protocol level message types, but
+ implements the authentication handshake via standard WAMP RPCs with well-known
+ procedure URIs and signatures.
+ """
+
+ URI_WAMP_BASE = "http://api.wamp.ws/"
+ """
+ WAMP base URI for WAMP predefined things.
+ """
+
+ URI_WAMP_ERROR = URI_WAMP_BASE + "error#"
+ """
+ Prefix for WAMP errors.
+ """
+
+ URI_WAMP_RPC = URI_WAMP_BASE + "procedure#"
+ """
+ Prefix for WAMP predefined RPCs.
+ """
+
+ URI_WAMP_EVENT = URI_WAMP_BASE + "event#"
+ """
+ Prefix for WAMP predefined PubSub events.
+ """
+
+
+class WampCraClientProtocol(WampClientProtocol, WampCraProtocol):
+ """
+ Simple, authenticated WAMP client protocol.
+
+ The client can perform WAMP-Challenge-Response-Authentication ("WAMP-CRA") to authenticate
+ itself to a WAMP server. The server needs to implement WAMP-CRA also of course.
+ """
+
+ def authSignature(self, authChallenge, authSecret = None):
+ """
+ Compute the authentication signature from an authentication challenge and a secret.
+
+ :param authChallenge: The authentication challenge.
+ :type authChallenge: str
+ :param authSecret: The authentication secret.
+ :type authSecret: str
+ :returns str -- The authentication signature.
+ """
+ if authSecret is None:
+ authSecret = ""
+ h = hmac.new(authSecret, authChallenge, hashlib.sha256)
+ sig = binascii.b2a_base64(h.digest()).strip()
+ return sig
+
+ def authenticate(self, onAuthSuccess, onAuthError = None, authKey = None, authExtra = None, authSecret = None):
+ """
+ Authenticate the WAMP session to server. Upon authentication success or failure, the appropriate callback will fire.
+
+ :param onAuthSuccess: Callback for authentication success.
+ :type onAuthSuccess: A callable.
+ :param onAuthError: Callback for authentication failure.
+ :type onAuthError: A callable.
+ :param authKey: The key of the authentication credentials, something like a user or application name.
+ :type authKey: str
+ :param authExtra: Any extra authentication information.
+ :type authExtra: dict
+ :param authSecret: The secret of the authentication credentials, something like the user password or application secret key.
+ """
+
+ def _onAuthError(e):
+ erroruri, errodesc, errordetails = e.value.args
+ if onAuthError is not None:
+ onAuthError(erroruri, errodesc, errordetails)
+
+ def _onAuthChallenge(challenge):
+ if authKey is not None:
+ sig = self.authSignature(challenge, authSecret)
+ else:
+ sig = None
+ d = self.call(WampCraProtocol.URI_WAMP_RPC + "auth", sig)
+ d.addCallbacks(onAuthSuccess, _onAuthError)
+
+ d = self.call(WampCraProtocol.URI_WAMP_RPC + "authreq", authKey, authExtra)
+ d.addCallbacks(_onAuthChallenge, _onAuthError)
+
+
+
+class WampCraServerProtocol(WampServerProtocol, WampCraProtocol):
+ """
+ Simple, authenticating WAMP server protocol.
+
+ The server lets clients perform WAMP-Challenge-Response-Authentication ("WAMP-CRA")
+ to authenticate. The clients need to implement WAMP-CRA also of course.
+
+ To implement an authenticating server, override:
+
+ * getAuthSecret
+ * getAuthPermissions
+ * onAuthenticated
+
+ in your class deriving from this class.
+ """
+
+ ## global client auth options
+ ##
+ clientAuthTimeout = 0
+ clientAuthAllowAnonymous = True
+
+
+ def getAuthPermissions(self, authKey, authExtra):
+ """
+ Get the permissions the session is granted when the authentication succeeds
+ for the given key / extra information.
+
+ Override in derived class to implement your authentication.
+
+ :param authKey: The authentication key.
+ :type authKey: str
+ :param authExtra: Authentication extra information.
+ :type authExtra: dict
+ :returns str -- The authentication secret for the key or None when the key does not exist.
+ """
+ return []
+
+
+ def getAuthSecret(self, authKey):
+ """
+ Get the authentication secret for an authentication key, i.e. the
+ user password for the user name. Return None when the authentication
+ key does not exist.
+
+ Override in derived class to implement your authentication.
+
+ :param authKey: The authentication key.
+ :type authKey: str
+ :returns str -- The authentication secret for the key or None when the key does not exist.
+ """
+ return None
+
+
+ def onAuthTimeout(self):
+ """
+ Fired when the client does not authenticate itself in time. The default implementation
+ will simply fail the connection.
+
+ May be overridden in derived class.
+ """
+ if not self.clientAuthenticated:
+ log.msg("failing connection upon client authentication timeout [%s secs]" % self.clientAuthTimeout)
+ self.failConnection()
+
+
+ def onAuthenticated(self, permissions):
+ """
+ Fired when client authentication was successful.
+
+ Override in derived class and register PubSub topics and/or RPC endpoints.
+
+ :param permissions: The permissions granted to the now authenticated client.
+ :type permissions: list
+ """
+ pass
+
+
+ def registerForPubSubFromPermissions(self, permissions):
+ """
+ Register topics for PubSub from auth permissions.
+
+ :param permissions: The permissions granted to the now authenticated client.
+ :type permissions: list
+ """
+ for p in permissions['pubsub']:
+ ## register topics for the clients
+ ##
+ pubsub = (WampServerProtocol.PUBLISH if p['pub'] else 0) | \
+ (WampServerProtocol.SUBSCRIBE if p['sub'] else 0)
+ topic = p['uri']
+ if self.pubHandlers.has_key(topic) or self.subHandlers.has_key(topic):
+ ## FIXME: handle dups!
+ log.msg("DUPLICATE TOPIC PERMISSION !!! " + topic)
+ self.registerForPubSub(topic, p['prefix'], pubsub)
+
+
+ def onSessionOpen(self):
+ """
+ Called when WAMP session has been established, but not yet authenticated. The default
+ implementation will prepare the session allowing the client to authenticate itself.
+ """
+
+ self.registerForRpc(self, WampCraProtocol.URI_WAMP_RPC, [WampCraServerProtocol.authRequest,
+ WampCraServerProtocol.auth])
+
+ ## reset authentication state
+ ##
+ self.clientAuthenticated = False
+ self.clientPendingAuth = None
+
+ ## client authentication timeout
+ ##
+ if self.clientAuthTimeout > 0:
+ self.clientAuthTimeoutCall = reactor.callLater(self.clientAuthTimeout, self.onAuthTimeout)
+ else:
+ self.clientAuthTimeoutCall = None
+
+
+ def authSignature(self, authChallenge, authKey = None):
+ """
+ Compute the authentication signature from an authentication challenge and for an authentication key.
+
+ :param authChallenge: The authentication challenge.
+ :type authChallenge: str
+ :param authKey: The authentication key for which to compute the signature.
+ :type authKey: str
+ :returns str -- The authentication signature.
+ """
+ if authKey is None:
+ secret = ""
+ else:
+ secret = self.getAuthSecret(authKey)
+ h = hmac.new(secret, authChallenge, hashlib.sha256)
+ sig = binascii.b2a_base64(h.digest()).strip()
+ return sig
+
+
+ @exportRpc("authreq")
+ def authRequest(self, appkey = None, extra = None):
+ """
+ RPC for clients to initiate the authentication handshake.
+
+ :param appkey: Authentication key, such as user name or application name.
+ :type appkey: str
+ :param extra: Authentication extra information.
+ :type extra: dict
+ :returns str -- Authentication challenge. The client will need to create an authentication signature from this.
+ """
+
+ ## check authentication state
+ ##
+ if self.clientAuthenticated:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "already-authenticated"), "already authenticated")
+ if self.clientPendingAuth is not None:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "authentication-already-requested"), "authentication request already issues - authentication pending")
+
+ ## check appkey
+ ##
+ if appkey is None and not self.clientAuthAllowAnonymous:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "anyonymous-auth-forbidden"), "authentication as anonymous forbidden")
+
+ if type(appkey) not in [str, unicode, types.NoneType]:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "invalid-argument"), "application key must be a string (was %s)" % str(type(appkey)))
+ if appkey is not None and self.getAuthSecret(appkey) is None:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "no-such-appkey"), "application key '%s' does not exist." % appkey)
+
+ ## check extra
+ ##
+ if extra:
+ if type(extra) != dict:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "invalid-argument"), "extra not a dictionary (was %s)." % str(type(extra)))
+ else:
+ extra = {}
+ for k in extra:
+ if type(extra[k]) not in [str, unicode, int, long, float, bool, types.NoneType]:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "invalid-argument"), "attribute '%s' in extra not a primitive type (was %s)" % (k, str(type(extra[k]))))
+
+ ## each authentication request gets a unique authid, which can only be used (later) once!
+ ##
+ authid = newid()
+
+ ## create authentication challenge
+ ##
+ info = {}
+ info['authid'] = authid
+ info['appkey'] = appkey
+ info['timestamp'] = utcnow()
+ info['sessionid'] = self.session_id
+ info['extra'] = extra
+
+ try:
+ pp = self.getAuthPermissions(appkey, extra)
+ if pp is None:
+ pp = {'pubsub': [], 'rpc': []}
+ info['permissions'] = pp
+ except Exception, e:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "auth-permissions-error"), str(e))
+
+ if appkey:
+ ## authenticated
+ ##
+ infoser = json.dumps(info)
+ sig = self.authSignature(infoser, appkey)
+
+ self.clientPendingAuth = (info, sig)
+ return infoser
+ else:
+ ## anonymous
+ ##
+ self.clientPendingAuth = (info, None)
+ return None
+
+
+ @exportRpc("auth")
+ def auth(self, signature = None):
+ """
+ RPC for clients to actually authenticate after requesting authentication and computing
+ a signature from the authentication challenge.
+
+ :param signature: Authenticatin signature computed by the client.
+ :type signature: str
+ :returns list -- A list of permissions the client is granted when authentication was successful.
+ """
+
+ ## check authentication state
+ ##
+ if self.clientAuthenticated:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "already-authenticated"), "already authenticated")
+ if self.clientPendingAuth is None:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "no-authentication-requested"), "no authentication previously requested")
+
+ ## check signature
+ ##
+ if type(signature) not in [str, unicode, types.NoneType]:
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "invalid-argument"), "signature must be a string or None (was %s)" % str(type(signature)))
+ if self.clientPendingAuth[1] != signature:
+ ## delete pending authentication, so that no retries are possible. authid is only valid for 1 try!!
+ ## FIXME: drop the connection?
+ self.clientPendingAuth = None
+ raise Exception(self.shrink(WampCraProtocol.URI_WAMP_ERROR + "invalid-signature"), "signature for authentication request is invalid")
+
+ ## at this point, the client has successfully authenticated!
+
+ ## get the permissions we determined earlier
+ ##
+ perms = self.clientPendingAuth[0]['permissions']
+
+ ## delete auth request and mark client as authenticated
+ ##
+ self.clientAppkey = self.clientPendingAuth[0]['appkey']
+ self.clientAuthenticated = True
+ self.clientPendingAuth = None
+ if self.clientAuthTimeoutCall is not None:
+ self.clientAuthTimeoutCall.cancel()
+ self.clientAuthTimeoutCall = None
+
+ ## fire authentication callback
+ ##
+ self.onAuthenticated(self.clientAppkey, perms)
+
+ ## return permissions to client
+ ##
+ return perms
+
diff --git a/autobahn/websocket.py b/autobahn/websocket.py new file mode 100644 index 0000000..1071ce6 --- /dev/null +++ b/autobahn/websocket.py @@ -0,0 +1,3636 @@ +###############################################################################
+##
+## Copyright 2011,2012 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+## The Python urlparse module currently does not contain the ws/wss
+## schemes, so we add those dynamically (which is a hack of course).
+##
+import urlparse
+wsschemes = ["ws", "wss"]
+urlparse.uses_relative.extend(wsschemes)
+urlparse.uses_netloc.extend(wsschemes)
+urlparse.uses_params.extend(wsschemes)
+urlparse.uses_query.extend(wsschemes)
+urlparse.uses_fragment.extend(wsschemes)
+
+from twisted.internet import reactor, protocol
+from twisted.python import log
+import urllib
+import binascii
+import hashlib
+import base64
+import struct
+import random
+import os
+from array import array
+from collections import deque
+from utf8validator import Utf8Validator
+from xormasker import XorMaskerNull, XorMaskerSimple, XorMaskerShifted1
+from httpstatus import *
+import autobahn # need autobahn.version
+
+
+def createWsUrl(hostname, port = None, isSecure = False, path = None, params = None):
+ """
+ Create a WbeSocket URL from components.
+
+ :param hostname: WebSocket server hostname.
+ :type hostname: str
+ :param port: WebSocket service port or None (to select default ports 80/443 depending on isSecure).
+ :type port: int
+ :param isSecure: Set True for secure WebSockets ("wss" scheme).
+ :type isSecure: bool
+ :param path: Path component of addressed resource (will be properly URL escaped).
+ :type path: str
+ :param params: A dictionary of key-values to construct the query component of the addressed resource (will be properly URL escaped).
+ :type params: dict
+
+ :returns str -- Constructed WebSocket URL.
+ """
+ if port is not None:
+ netloc = "%s:%d" % (hostname, port)
+ else:
+ if isSecure:
+ netloc = "%s:443" % hostname
+ else:
+ netloc = "%s:80" % hostname
+ if isSecure:
+ scheme = "wss"
+ else:
+ scheme = "ws"
+ if path is not None:
+ ppath = urllib.quote(path)
+ else:
+ ppath = "/"
+ if params is not None:
+ query = urllib.urlencode(params)
+ else:
+ query = None
+ return urlparse.urlunparse((scheme, netloc, ppath, None, query, None))
+
+
+def parseWsUrl(url):
+ """
+ Parses as WebSocket URL into it's components and returns a tuple (isSecure, host, port, resource, path, params).
+
+ isSecure is a flag which is True for wss URLs.
+ host is the hostname or IP from the URL.
+ port is the port from the URL or standard port derived from scheme (ws = 80, wss = 443).
+ resource is the /resource name/ from the URL, the /path/ together with the (optional) /query/ component.
+ path is the /path/ component properly unescaped.
+ params is the /query) component properly unescaped and returned as dictionary.
+
+ :param url: A valid WebSocket URL, i.e. ws://localhost:9000/myresource?param1=23¶m2=666
+ :type url: str
+
+ :returns: tuple -- A tuple (isSecure, host, port, resource, path, params)
+ """
+ parsed = urlparse.urlparse(url)
+ if parsed.scheme not in ["ws", "wss"]:
+ raise Exception("invalid WebSocket scheme '%s'" % parsed.scheme)
+ if parsed.port is None or parsed.port == "":
+ if parsed.scheme == "ws":
+ port = 80
+ else:
+ port = 443
+ else:
+ port = int(parsed.port)
+ if parsed.fragment is not None and parsed.fragment != "":
+ raise Exception("invalid WebSocket URL: non-empty fragment '%s" % parsed.fragment)
+ if parsed.path is not None and parsed.path != "":
+ ppath = parsed.path
+ path = urllib.unquote(ppath)
+ else:
+ ppath = "/"
+ path = ppath
+ if parsed.query is not None and parsed.query != "":
+ resource = ppath + "?" + parsed.query
+ params = urlparse.parse_qs(parsed.query)
+ else:
+ resource = ppath
+ params = {}
+ return (parsed.scheme == "wss", parsed.hostname, port, resource, path, params)
+
+
+def connectWS(factory, contextFactory = None, timeout = 30, bindAddress = None):
+ """
+ Establish WebSockets connection to a server. The connection parameters like target
+ host, port, resource and others are provided via the factory.
+
+ :param factory: The WebSockets protocol factory to be used for creating client protocol instances.
+ :type factory: An :class:`autobahn.websocket.WebSocketClientFactory` instance.
+ :param contextFactory: SSL context factory, required for secure WebSockets connections ("wss").
+ :type contextFactory: A twisted.internet.ssl.ClientContextFactory instance.
+ :param timeout: Number of seconds to wait before assuming the connection has failed.
+ :type timeout: int
+ :param bindAddress: A (host, port) tuple of local address to bind to, or None.
+ :type bindAddress: tuple
+
+ :returns: obj -- An object which provides twisted.interface.IConnector.
+ """
+ if factory.isSecure:
+ if contextFactory is None:
+ # create default client SSL context factory when none given
+ from twisted.internet import ssl
+ contextFactory = ssl.ClientContextFactory()
+ conn = reactor.connectSSL(factory.host, factory.port, factory, contextFactory, timeout, bindAddress)
+ else:
+ conn = reactor.connectTCP(factory.host, factory.port, factory, timeout, bindAddress)
+ return conn
+
+
+def listenWS(factory, contextFactory = None, backlog = 50, interface = ''):
+ """
+ Listen for incoming WebSocket connections from clients. The connection parameters like
+ listening port and others are provided via the factory.
+
+ :param factory: The WebSockets protocol factory to be used for creating server protocol instances.
+ :type factory: An :class:`autobahn.websocket.WebSocketServerFactory` instance.
+ :param contextFactory: SSL context factory, required for secure WebSockets connections ("wss").
+ :type contextFactory: A twisted.internet.ssl.ContextFactory.
+ :param backlog: Size of the listen queue.
+ :type backlog: int
+ :param interface: The interface (derived from hostname given) to bind to, defaults to '' (all).
+ :type interface: str
+
+ :returns: obj -- An object that provides twisted.interface.IListeningPort.
+ """
+ if factory.isSecure:
+ if contextFactory is None:
+ raise Exception("Secure WebSocket listen requested, but no SSL context factory given")
+ listener = reactor.listenSSL(factory.port, factory, contextFactory, backlog, interface)
+ else:
+ listener = reactor.listenTCP(factory.port, factory, backlog, interface)
+ return listener
+
+
+class FrameHeader:
+ """
+ Thin-wrapper for storing WebSockets frame metadata.
+
+ FOR INTERNAL USE ONLY!
+ """
+
+ def __init__(self, opcode, fin, rsv, length, mask):
+ """
+ Constructor.
+
+ :param opcode: Frame opcode (0-15).
+ :type opcode: int
+ :param fin: Frame FIN flag.
+ :type fin: bool
+ :param rsv: Frame reserved flags (0-7).
+ :type rsv: int
+ :param length: Frame payload length.
+ :type length: int
+ :param mask: Frame mask (binary string) or None.
+ :type mask: str
+ """
+ self.opcode = opcode
+ self.fin = fin
+ self.rsv = rsv
+ self.length = length
+ self.mask = mask
+
+
+class HttpException():
+ """
+ Throw an instance of this class to deny a WebSockets connection
+ during handshake in :meth:`autobahn.websocket.WebSocketServerProtocol.onConnect`.
+ You can find definitions of HTTP status codes in module :mod:`autobahn.httpstatus`.
+ """
+
+ def __init__(self, code, reason):
+ """
+ Constructor.
+
+ :param code: HTTP error code.
+ :type code: int
+ :param reason: HTTP error reason.
+ :type reason: str
+ """
+ self.code = code
+ self.reason = reason
+
+
+class ConnectionRequest():
+ """
+ Thin-wrapper for WebSockets connection request information
+ provided in :meth:`autobahn.websocket.WebSocketServerProtocol.onConnect` when a WebSockets
+ client establishes a connection to a WebSockets server.
+ """
+ def __init__(self, peer, peerstr, headers, host, path, params, version, origin, protocols, extensions):
+ """
+ Constructor.
+
+ :param peer: IP address/port of the connecting client.
+ :type peer: object
+ :param peerstr: IP address/port of the connecting client as string.
+ :type peerstr: str
+ :param headers: HTTP headers from opening handshake request.
+ :type headers: dict
+ :param host: Host from opening handshake HTTP header.
+ :type host: str
+ :param path: Path from requested HTTP resource URI. For example, a resource URI of "/myservice?foo=23&foo=66&bar=2" will be parsed to "/myservice".
+ :type path: str
+ :param params: Query parameters (if any) from requested HTTP resource URI. For example, a resource URI of "/myservice?foo=23&foo=66&bar=2" will be parsed to {'foo': ['23', '66'], 'bar': ['2']}.
+ :type params: dict of arrays of strings
+ :param version: The WebSockets protocol version the client announced (and will be spoken, when connection is accepted).
+ :type version: int
+ :param origin: The WebSockets origin header or None. Note that this only a reliable source of information for browser clients!
+ :type origin: str
+ :param protocols: The WebSockets (sub)protocols the client announced. You must select and return one of those (or None) in :meth:`autobahn.websocket.WebSocketServerProtocol.onConnect`.
+ :type protocols: array of strings
+ :param extensions: The WebSockets extensions the client requested and the server accepted (and thus will be spoken, when WS connection is established).
+ :type extensions: array of strings
+ """
+ self.peer = peer
+ self.peerstr = peerstr
+ self.headers = headers
+ self.host = host
+ self.path = path
+ self.params = params
+ self.version = version
+ self.origin = origin
+ self.protocols = protocols
+ self.extensions = extensions
+
+
+class ConnectionResponse():
+ """
+ Thin-wrapper for WebSockets connection response information
+ provided in :meth:`autobahn.websocket.WebSocketClientProtocol.onConnect` when a WebSockets
+ client has established a connection to a WebSockets server.
+ """
+ def __init__(self, peer, peerstr, headers, version, protocol, extensions):
+ """
+ Constructor.
+
+ :param peer: IP address/port of the connected server.
+ :type peer: object
+ :param peerstr: IP address/port of the connected server as string.
+ :type peerstr: str
+ :param headers: HTTP headers from opening handshake response.
+ :type headers: dict
+ :param version: The WebSockets protocol version that is spoken.
+ :type version: int
+ :param protocol: The WebSockets (sub)protocol in use.
+ :type protocol: str
+ :param extensions: The WebSockets extensions in use.
+ :type extensions: array of strings
+ """
+ self.peer = peer
+ self.peerstr = peerstr
+ self.headers = headers
+ self.version = version
+ self.protocol = protocol
+ self.extensions = extensions
+
+
+def parseHttpHeader(data):
+ """
+ Parses the beginning of a HTTP request header (the data up to the \n\n line) into a pair
+ of status line and HTTP headers dictionary.
+ Header keys are normalized to all-lower-case.
+
+ FOR INTERNAL USE ONLY!
+
+ :param data: The HTTP header data up to the \n\n line.
+ :type data: str
+ """
+ raw = data.splitlines()
+ http_status_line = raw[0].strip()
+ http_headers = {}
+ http_headers_cnt = {}
+ for h in raw[1:]:
+ i = h.find(":")
+ if i > 0:
+ ## HTTP header keys are case-insensitive
+ key = h[:i].strip().lower()
+
+ ## not sure if UTF-8 is allowed for HTTP header values..
+ value = h[i+1:].strip().decode("utf-8")
+
+ ## handle HTTP headers split across multiple lines
+ if http_headers.has_key(key):
+ http_headers[key] += ", %s" % value
+ http_headers_cnt[key] += 1
+ else:
+ http_headers[key] = value
+ http_headers_cnt[key] = 1
+ else:
+ # skip bad HTTP header
+ pass
+ return (http_status_line, http_headers, http_headers_cnt)
+
+
+class WebSocketProtocol(protocol.Protocol):
+ """
+ A Twisted Protocol class for WebSockets. This class is used by both WebSocket
+ client and server protocol version. It is unusable standalone, for example
+ the WebSockets initial handshake is implemented in derived class differently
+ for clients and servers.
+ """
+
+ SUPPORTED_SPEC_VERSIONS = [0, 10, 11, 12, 13, 14, 15, 16, 17, 18]
+ """
+ WebSockets protocol spec (draft) versions supported by this implementation.
+ Use of version 18 indicates RFC6455. Use of versions < 18 indicate actual
+ draft spec versions (Hybi-Drafts). Use of version 0 indicates Hixie-76.
+ """
+
+ SUPPORTED_PROTOCOL_VERSIONS = [0, 8, 13]
+ """
+ WebSocket protocol versions supported by this implementation. For Hixie-76,
+ there is no protocol version announced in HTTP header, and we just use the
+ draft version (0) in this case.
+ """
+
+ SPEC_TO_PROTOCOL_VERSION = {0: 0, 10: 8, 11: 8, 12: 8, 13: 13, 14: 13, 15: 13, 16: 13, 17: 13, 18: 13}
+ """
+ Mapping from protocol spec (draft) version to protocol version. For Hixie-76,
+ there is no protocol version announced in HTTP header, and we just use the
+ pseudo protocol version 0 in this case.
+ """
+
+ PROTOCOL_TO_SPEC_VERSION = {0: 0, 8: 12, 13: 18}
+ """
+ Mapping from protocol version to the latest protocol spec (draft) version
+ using that protocol version. For Hixie-76, there is no protocol version
+ announced in HTTP header, and we just use the draft version (0) in this case.
+ """
+
+ DEFAULT_SPEC_VERSION = 10
+ """
+ Default WebSockets protocol spec version this implementation speaks.
+ We use Hybi-10, since this is what is currently targeted by widely distributed
+ browsers (namely Firefox 8 and the like).
+ """
+
+ DEFAULT_ALLOW_HIXIE76 = False
+ """
+ By default, this implementation will not allow to speak the obsoleted
+ Hixie-76 protocol version. That protocol version has security issues, but
+ is still spoken by some clients. Enable at your own risk! Enabling can be
+ done by using setProtocolOptions() on the factories for clients and servers.
+ """
+
+ WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+ """
+ Protocol defined magic used during WebSocket handshake (used in Hybi-drafts
+ and final RFC6455.
+ """
+
+ QUEUED_WRITE_DELAY = 0.00001
+ """For synched/chopped writes, this is the reactor reentry delay in seconds."""
+
+ PAYLOAD_LEN_XOR_BREAKEVEN = 128
+ """Tuning parameter which chooses XORer used for masking/unmasking based on
+ payload length."""
+
+ MESSAGE_TYPE_TEXT = 1
+ """WebSockets text message type (UTF-8 payload)."""
+
+ MESSAGE_TYPE_BINARY = 2
+ """WebSockets binary message type (arbitrary binary payload)."""
+
+ ## WebSockets protocol state:
+ ## STATE_CONNECTING => STATE_OPEN => STATE_CLOSING => STATE_CLOSED
+ ##
+ STATE_CLOSED = 0
+ STATE_CONNECTING = 1
+ STATE_CLOSING = 2
+ STATE_OPEN = 3
+
+ ## Streaming Send State
+ SEND_STATE_GROUND = 0
+ SEND_STATE_MESSAGE_BEGIN = 1
+ SEND_STATE_INSIDE_MESSAGE = 2
+ SEND_STATE_INSIDE_MESSAGE_FRAME = 3
+
+ ## WebSockets protocol close codes
+ ##
+ CLOSE_STATUS_CODE_NORMAL = 1000
+ """Normal close of connection."""
+
+ CLOSE_STATUS_CODE_GOING_AWAY = 1001
+ """Going away."""
+
+ CLOSE_STATUS_CODE_PROTOCOL_ERROR = 1002
+ """Protocol error."""
+
+ CLOSE_STATUS_CODE_UNSUPPORTED_DATA = 1003
+ """Unsupported data."""
+
+ CLOSE_STATUS_CODE_RESERVED1 = 1004
+ """RESERVED"""
+
+ CLOSE_STATUS_CODE_NULL = 1005 # MUST NOT be set in close frame!
+ """No status received. (MUST NOT be used as status code when sending a close)."""
+
+ CLOSE_STATUS_CODE_ABNORMAL_CLOSE = 1006 # MUST NOT be set in close frame!
+ """Abnormal close of connection. (MUST NOT be used as status code when sending a close)."""
+
+ CLOSE_STATUS_CODE_INVALID_PAYLOAD = 1007
+ """Invalid frame payload data."""
+
+ CLOSE_STATUS_CODE_POLICY_VIOLATION = 1008
+ """Policy violation."""
+
+ CLOSE_STATUS_CODE_MESSAGE_TOO_BIG = 1009
+ """Message too big."""
+
+ CLOSE_STATUS_CODE_MANDATORY_EXTENSION = 1010
+ """Mandatory extension."""
+
+ CLOSE_STATUS_CODE_INTERNAL_ERROR = 1011
+ """The peer encountered an unexpected condition or internal error."""
+
+ CLOSE_STATUS_CODE_TLS_HANDSHAKE_FAILED = 1015 # MUST NOT be set in close frame!
+ """TLS handshake failed, i.e. server certificate could not be verified. (MUST NOT be used as status code when sending a close)."""
+
+ CLOSE_STATUS_CODES_ALLOWED = [CLOSE_STATUS_CODE_NORMAL,
+ CLOSE_STATUS_CODE_GOING_AWAY,
+ CLOSE_STATUS_CODE_PROTOCOL_ERROR,
+ CLOSE_STATUS_CODE_UNSUPPORTED_DATA,
+ CLOSE_STATUS_CODE_INVALID_PAYLOAD,
+ CLOSE_STATUS_CODE_POLICY_VIOLATION,
+ CLOSE_STATUS_CODE_MESSAGE_TOO_BIG,
+ CLOSE_STATUS_CODE_MANDATORY_EXTENSION,
+ CLOSE_STATUS_CODE_INTERNAL_ERROR]
+ """Status codes allowed to send in close."""
+
+
+ def onOpen(self):
+ """
+ Callback when initial WebSockets handshake was completed. Now you may send messages.
+ Default implementation does nothing. Override in derived class.
+
+ Modes: Hybi, Hixie
+ """
+ if self.debugCodePaths:
+ log.msg("WebSocketProtocol.onOpen")
+
+
+ def onMessageBegin(self, opcode):
+ """
+ Callback when receiving a new message has begun. Default implementation will
+ prepare to buffer message frames. Override in derived class.
+
+ Modes: Hybi, Hixie
+
+ :param opcode: Opcode of message.
+ :type opcode: int
+ """
+ self.message_opcode = opcode
+ self.message_data = []
+ self.message_data_total_length = 0
+
+
+ def onMessageFrameBegin(self, length, reserved):
+ """
+ Callback when receiving a new message frame has begun. Default implementation will
+ prepare to buffer message frame data. Override in derived class.
+
+ Modes: Hybi
+
+ :param length: Payload length of message frame which is to be received.
+ :type length: int
+ :param reserved: Reserved bits set in frame (an integer from 0 to 7).
+ :type reserved: int
+ """
+ self.frame_length = length
+ self.frame_reserved = reserved
+ self.frame_data = []
+ self.message_data_total_length += length
+ if not self.failedByMe:
+ if self.maxMessagePayloadSize > 0 and self.message_data_total_length > self.maxMessagePayloadSize:
+ self.wasMaxMessagePayloadSizeExceeded = True
+ self.failConnection(WebSocketProtocol.CLOSE_STATUS_CODE_MESSAGE_TOO_BIG, "message exceeds payload limit of %d octets" % self.maxMessagePayloadSize)
+ elif self.maxFramePayloadSize > 0 and length > self.maxFramePayloadSize:
+ self.wasMaxFramePayloadSizeExceeded = True
+ self.failConnection(WebSocketProtocol.CLOSE_STATUS_CODE_POLICY_VIOLATION, "frame exceeds payload limit of %d octets" % self.maxFramePayloadSize)
+
+
+ def onMessageFrameData(self, payload):
+ """
+ Callback when receiving data witin message frame. Default implementation will
+ buffer data for frame. Override in derived class.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, this method is slightly misnamed for historic reasons.
+
+ :param payload: Partial payload for message frame.
+ :type payload: str
+ """
+ if not self.failedByMe:
+ if self.websocket_version == 0:
+ self.message_data_total_length += len(payload)
+ if self.maxMessagePayloadSize > 0 and self.message_data_total_length > self.maxMessagePayloadSize:
+ self.wasMaxMessagePayloadSizeExceeded = True
+ self.failConnection(WebSocketProtocol.CLOSE_STATUS_CODE_MESSAGE_TOO_BIG, "message exceeds payload limit of %d octets" % self.maxMessagePayloadSize)
+ self.message_data.append(payload)
+ else:
+ self.frame_data.append(payload)
+
+
+ def onMessageFrameEnd(self):
+ """
+ Callback when a message frame has been completely received. Default implementation
+ will flatten the buffered frame data and callback onMessageFrame. Override
+ in derived class.
+
+ Modes: Hybi
+ """
+ if not self.failedByMe:
+ self.onMessageFrame(self.frame_data, self.frame_reserved)
+
+ self.frame_data = None
+
+
+ def onMessageFrame(self, payload, reserved):
+ """
+ Callback fired when complete message frame has been received. Default implementation
+ will buffer frame for message. Override in derived class.
+
+ Modes: Hybi
+
+ :param payload: Message frame payload.
+ :type payload: list of str
+ :param reserved: Reserved bits set in frame (an integer from 0 to 7).
+ :type reserved: int
+ """
+ if not self.failedByMe:
+ self.message_data.extend(payload)
+
+
+ def onMessageEnd(self):
+ """
+ Callback when a message has been completely received. Default implementation
+ will flatten the buffered frames and callback onMessage. Override
+ in derived class.
+
+ Modes: Hybi, Hixie
+ """
+ if not self.failedByMe:
+ payload = ''.join(self.message_data)
+ self.onMessage(payload, self.message_opcode == WebSocketProtocol.MESSAGE_TYPE_BINARY)
+
+ self.message_data = None
+
+
+ def onMessage(self, payload, binary):
+ """
+ Callback when a complete message was received. Default implementation does nothing.
+ Override in derived class.
+
+ Modes: Hybi, Hixie
+
+ :param payload: Message payload (UTF-8 encoded text string or binary string). Can also be an empty string, when message contained no payload.
+ :type payload: str
+ :param binary: If True, payload is binary, otherwise text.
+ :type binary: bool
+ """
+ if self.debug:
+ log.msg("WebSocketProtocol.onMessage")
+
+
+ def onPing(self, payload):
+ """
+ Callback when Ping was received. Default implementation responds
+ with a Pong. Override in derived class.
+
+ Modes: Hybi
+
+ :param payload: Payload of Ping, when there was any. Can be arbitrary, up to 125 octets.
+ :type payload: str
+ """
+ if self.debug:
+ log.msg("WebSocketProtocol.onPing")
+ if self.state == WebSocketProtocol.STATE_OPEN:
+ self.sendPong(payload)
+
+
+ def onPong(self, payload):
+ """
+ Callback when Pong was received. Default implementation does nothing.
+ Override in derived class.
+
+ Modes: Hybi
+
+ :param payload: Payload of Pong, when there was any. Can be arbitrary, up to 125 octets.
+ """
+ if self.debug:
+ log.msg("WebSocketProtocol.onPong")
+
+
+ def onClose(self, wasClean, code, reason):
+ """
+ Callback when the connection has been closed. Override in derived class.
+
+ Modes: Hybi, Hixie
+
+ :param wasClean: True, iff the connection was closed cleanly.
+ :type wasClean: bool
+ :param code: None or close status code (sent by peer), if there was one (:class:`WebSocketProtocol`.CLOSE_STATUS_CODE_*).
+ :type code: int
+ :param reason: None or close reason (sent by peer) (when present, a status code MUST have been also be present).
+ :type reason: str
+ """
+ if self.debugCodePaths:
+ s = "WebSocketProtocol.onClose:\n"
+ s += "wasClean=%s\n" % wasClean
+ s += "code=%s\n" % code
+ s += "reason=%s\n" % reason
+ s += "self.closedByMe=%s\n" % self.closedByMe
+ s += "self.failedByMe=%s\n" % self.failedByMe
+ s += "self.droppedByMe=%s\n" % self.droppedByMe
+ s += "self.wasClean=%s\n" % self.wasClean
+ s += "self.wasNotCleanReason=%s\n" % self.wasNotCleanReason
+ s += "self.localCloseCode=%s\n" % self.localCloseCode
+ s += "self.localCloseReason=%s\n" % self.localCloseReason
+ s += "self.remoteCloseCode=%s\n" % self.remoteCloseCode
+ s += "self.remoteCloseReason=%s\n" % self.remoteCloseReason
+ log.msg(s)
+
+
+ def onCloseFrame(self, code, reasonRaw):
+ """
+ Callback when a Close frame was received. The default implementation answers by
+ sending a Close when no Close was sent before. Otherwise it drops
+ the TCP connection either immediately (when we are a server) or after a timeout
+ (when we are a client and expect the server to drop the TCP).
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, this method is slightly misnamed for historic reasons.
+ - For Hixie mode, code and reasonRaw are silently ignored.
+
+ :param code: None or close status code, if there was one (:class:`WebSocketProtocol`.CLOSE_STATUS_CODE_*).
+ :type code: int
+ :param reason: None or close reason (when present, a status code MUST have been also be present).
+ :type reason: str
+ """
+ if self.debugCodePaths:
+ log.msg("WebSocketProtocol.onCloseFrame")
+
+ self.remoteCloseCode = code
+ self.remoteCloseReason = reasonRaw
+
+ ## reserved close codes: 0-999, 1004, 1005, 1006, 1011-2999, >= 5000
+ ##
+ if code is not None and (code < 1000 or (code >= 1000 and code <= 2999 and code not in WebSocketProtocol.CLOSE_STATUS_CODES_ALLOWED) or code >= 5000):
+ if self.protocolViolation("invalid close code %d" % code):
+ return True
+
+ ## closing reason
+ ##
+ if reasonRaw is not None:
+ ## we use our own UTF-8 validator to get consistent and fully conformant
+ ## UTF-8 validation behavior
+ u = Utf8Validator()
+ val = u.validate(reasonRaw)
+ if not val[0]:
+ if self.invalidPayload("invalid close reason (non-UTF-8 payload)"):
+ return True
+
+ if self.state == WebSocketProtocol.STATE_CLOSING:
+ ## We already initiated the closing handshake, so this
+ ## is the peer's reply to our close frame.
+
+ ## cancel any closing HS timer if present
+ ##
+ if self.closeHandshakeTimeoutCall is not None:
+ if self.debugCodePaths:
+ log.msg("closeHandshakeTimeoutCall.cancel")
+ self.closeHandshakeTimeoutCall.cancel()
+ self.closeHandshakeTimeoutCall = None
+
+ self.wasClean = True
+
+ if self.isServer:
+ ## When we are a server, we immediately drop the TCP.
+ self.dropConnection(abort = True)
+ else:
+ ## When we are a client, the server should drop the TCP
+ ## If that doesn't happen, we do. And that will set wasClean = False.
+ if self.serverConnectionDropTimeout > 0:
+ self.serverConnectionDropTimeoutCall = reactor.callLater(self.serverConnectionDropTimeout, self.onServerConnectionDropTimeout)
+
+ elif self.state == WebSocketProtocol.STATE_OPEN:
+ ## The peer initiates a closing handshake, so we reply
+ ## by sending close frame.
+
+ self.wasClean = True
+
+ if self.websocket_version == 0:
+ self.sendCloseFrame(isReply = True)
+ else:
+ ## Either reply with same code/reason, or code == NORMAL/reason=None
+ if self.echoCloseCodeReason:
+ self.sendCloseFrame(code = code, reasonUtf8 = reason.encode("UTF-8"), isReply = True)
+ else:
+ self.sendCloseFrame(code = WebSocketProtocol.CLOSE_STATUS_CODE_NORMAL, isReply = True)
+
+ if self.isServer:
+ ## When we are a server, we immediately drop the TCP.
+ self.dropConnection(abort = False)
+ else:
+ ## When we are a client, we expect the server to drop the TCP,
+ ## and when the server fails to do so, a timeout in sendCloseFrame()
+ ## will set wasClean = False back again.
+ pass
+
+ else:
+ ## STATE_CONNECTING, STATE_CLOSED
+ raise Exception("logic error")
+
+
+ def onServerConnectionDropTimeout(self):
+ """
+ We (a client) expected the peer (a server) to drop the connection,
+ but it didn't (in time self.serverConnectionDropTimeout).
+ So we drop the connection, but set self.wasClean = False.
+
+ Modes: Hybi, Hixie
+ """
+ self.serverConnectionDropTimeoutCall = None
+ if self.state != WebSocketProtocol.STATE_CLOSED:
+ if self.debugCodePaths:
+ log.msg("onServerConnectionDropTimeout")
+ self.wasClean = False
+ self.wasNotCleanReason = "server did not drop TCP connection (in time)"
+ self.wasServerConnectionDropTimeout = True
+ self.dropConnection(abort = True)
+ else:
+ if self.debugCodePaths:
+ log.msg("skipping onServerConnectionDropTimeout since connection is already closed")
+
+
+ def onOpenHandshakeTimeout(self):
+ """
+ We expected the peer to complete the opening handshake with to us.
+ It didn't do so (in time self.openHandshakeTimeout).
+ So we drop the connection, but set self.wasClean = False.
+
+ Modes: Hybi, Hixie
+ """
+ self.openHandshakeTimeoutCall = None
+ if self.state == WebSocketProtocol.STATE_CONNECTING:
+ if self.debugCodePaths:
+ log.msg("onOpenHandshakeTimeout fired")
+ self.wasClean = False
+ self.wasNotCleanReason = "peer did not finish (in time) the opening handshake"
+ self.wasOpenHandshakeTimeout = True
+ self.dropConnection(abort = True)
+ elif self.state == WebSocketProtocol.STATE_OPEN:
+ if self.debugCodePaths:
+ log.msg("skipping onOpenHandshakeTimeout since WebSocket connection is open (opening handshake already finished)")
+ elif self.state == WebSocketProtocol.STATE_CLOSING:
+ if self.debugCodePaths:
+ log.msg("skipping onOpenHandshakeTimeout since WebSocket connection is closing")
+ elif self.state == WebSocketProtocol.STATE_CLOSED:
+ if self.debugCodePaths:
+ log.msg("skipping onOpenHandshakeTimeout since WebSocket connection already closed")
+ else:
+ # should not arrive here
+ raise Exception("logic error")
+
+
+ def onCloseHandshakeTimeout(self):
+ """
+ We expected the peer to respond to us initiating a close handshake. It didn't
+ respond (in time self.closeHandshakeTimeout) with a close response frame though.
+ So we drop the connection, but set self.wasClean = False.
+
+ Modes: Hybi, Hixie
+ """
+ self.closeHandshakeTimeoutCall = None
+ if self.state != WebSocketProtocol.STATE_CLOSED:
+ if self.debugCodePaths:
+ log.msg("onCloseHandshakeTimeout fired")
+ self.wasClean = False
+ self.wasNotCleanReason = "peer did not respond (in time) in closing handshake"
+ self.wasCloseHandshakeTimeout = True
+ self.dropConnection(abort = True)
+ else:
+ if self.debugCodePaths:
+ log.msg("skipping onCloseHandshakeTimeout since connection is already closed")
+
+
+ def dropConnection(self, abort = False):
+ """
+ Drop the underlying TCP connection. For abort parameter, see:
+
+ * http://twistedmatrix.com/documents/current/core/howto/servers.html#auto2
+ * https://github.com/tavendo/AutobahnPython/issues/96
+
+ Modes: Hybi, Hixie
+ """
+ if self.state != WebSocketProtocol.STATE_CLOSED:
+ if self.debugCodePaths:
+ log.msg("dropping connection")
+ self.droppedByMe = True
+ self.state = WebSocketProtocol.STATE_CLOSED
+
+ if abort:
+ self.transport.abortConnection()
+ else:
+ self.transport.loseConnection()
+ else:
+ if self.debugCodePaths:
+ log.msg("skipping dropConnection since connection is already closed")
+
+
+ def failConnection(self, code = CLOSE_STATUS_CODE_GOING_AWAY, reason = "Going Away"):
+ """
+ Fails the WebSockets connection.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, the code and reason are silently ignored.
+ """
+ if self.state != WebSocketProtocol.STATE_CLOSED:
+ if self.debugCodePaths:
+ log.msg("Failing connection : %s - %s" % (code, reason))
+ self.failedByMe = True
+ if self.failByDrop:
+ ## brutally drop the TCP connection
+ self.wasClean = False
+ self.wasNotCleanReason = "I failed the WebSocket connection by dropping the TCP connection"
+ self.dropConnection(abort = True)
+ else:
+ ## perform WebSockets closing handshake
+ if self.state != WebSocketProtocol.STATE_CLOSING:
+ self.sendCloseFrame(code = code, reasonUtf8 = reason.encode("UTF-8"), isReply = False)
+ else:
+ if self.debugCodePaths:
+ log.msg("skipping failConnection since connection is already closing")
+ else:
+ if self.debugCodePaths:
+ log.msg("skipping failConnection since connection is already closed")
+
+
+ def protocolViolation(self, reason):
+ """
+ Fired when a WebSockets protocol violation/error occurs.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, reason is silently ignored.
+
+ :param reason: Protocol violation that was encountered (human readable).
+ :type reason: str
+
+ :returns: bool -- True, when any further processing should be discontinued.
+ """
+ if self.debugCodePaths:
+ log.msg("Protocol violation : %s" % reason)
+ self.failConnection(WebSocketProtocol.CLOSE_STATUS_CODE_PROTOCOL_ERROR, reason)
+ if self.failByDrop:
+ return True
+ else:
+ ## if we don't immediately drop the TCP, we need to skip the invalid frame
+ ## to continue to later receive the closing handshake reply
+ return False
+
+
+ def invalidPayload(self, reason):
+ """
+ Fired when invalid payload is encountered. Currently, this only happens
+ for text message when payload is invalid UTF-8 or close frames with
+ close reason that is invalid UTF-8.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, reason is silently ignored.
+
+ :param reason: What was invalid for the payload (human readable).
+ :type reason: str
+
+ :returns: bool -- True, when any further processing should be discontinued.
+ """
+ if self.debugCodePaths:
+ log.msg("Invalid payload : %s" % reason)
+ self.failConnection(WebSocketProtocol.CLOSE_STATUS_CODE_INVALID_PAYLOAD, reason)
+ if self.failByDrop:
+ return True
+ else:
+ ## if we don't immediately drop the TCP, we need to skip the invalid frame
+ ## to continue to later receive the closing handshake reply
+ return False
+
+
+ def connectionMade(self):
+ """
+ This is called by Twisted framework when a new TCP connection has been established
+ and handed over to a Protocol instance (an instance of this class).
+
+ Modes: Hybi, Hixie
+ """
+
+ ## copy default options from factory (so we are not affected by changed on those)
+ ##
+
+ self.debug = self.factory.debug
+ self.debugCodePaths = self.factory.debugCodePaths
+
+ self.logOctets = self.factory.logOctets
+ self.logFrames = self.factory.logFrames
+
+ self.allowHixie76 = self.factory.allowHixie76
+ self.utf8validateIncoming = self.factory.utf8validateIncoming
+ self.applyMask = self.factory.applyMask
+ self.maxFramePayloadSize = self.factory.maxFramePayloadSize
+ self.maxMessagePayloadSize = self.factory.maxMessagePayloadSize
+ self.autoFragmentSize = self.factory.autoFragmentSize
+ self.failByDrop = self.factory.failByDrop
+ self.echoCloseCodeReason = self.factory.echoCloseCodeReason
+ self.openHandshakeTimeout = self.factory.openHandshakeTimeout
+ self.closeHandshakeTimeout = self.factory.closeHandshakeTimeout
+ self.tcpNoDelay = self.factory.tcpNoDelay
+
+ if self.isServer:
+ self.versions = self.factory.versions
+ self.webStatus = self.factory.webStatus
+ self.requireMaskedClientFrames = self.factory.requireMaskedClientFrames
+ self.maskServerFrames = self.factory.maskServerFrames
+ else:
+ self.version = self.factory.version
+ self.acceptMaskedServerFrames = self.factory.acceptMaskedServerFrames
+ self.maskClientFrames = self.factory.maskClientFrames
+ self.serverConnectionDropTimeout = self.factory.serverConnectionDropTimeout
+
+ ## Set "Nagle"
+ self.transport.setTcpNoDelay(self.tcpNoDelay)
+
+ ## the peer we are connected to
+ self.peer = self.transport.getPeer()
+ self.peerstr = "%s:%d" % (self.peer.host, self.peer.port)
+
+ ## initial state
+ self.state = WebSocketProtocol.STATE_CONNECTING
+ self.send_state = WebSocketProtocol.SEND_STATE_GROUND
+ self.data = ""
+
+ ## for chopped/synched sends, we need to queue to maintain
+ ## ordering when recalling the reactor to actually "force"
+ ## the octets to wire (see test/trickling in the repo)
+ self.send_queue = deque()
+ self.triggered = False
+
+ ## incremental UTF8 validator
+ self.utf8validator = Utf8Validator()
+
+ ## track when frame/message payload sizes (incoming) were exceeded
+ self.wasMaxFramePayloadSizeExceeded = False
+ self.wasMaxMessagePayloadSizeExceeded = False
+
+ ## the following vars are related to connection close handling/tracking
+
+ # True, iff I have initiated closing HS (that is, did send close first)
+ self.closedByMe = False
+
+ # True, iff I have failed the WS connection (i.e. due to protocol error)
+ # Failing can be either by initiating close HS or brutal drop (this is
+ # controlled by failByDrop option)
+ self.failedByMe = False
+
+ # True, iff I dropped the TCP connection (called transport.loseConnection())
+ self.droppedByMe = False
+
+ # True, iff full WebSockets closing handshake was performed (close frame sent
+ # and received) _and_ the server dropped the TCP (which is its responsibility)
+ self.wasClean = False
+
+ # When self.wasClean = False, the reason (what happened)
+ self.wasNotCleanReason = None
+
+ # When we are a client, and we expected the server to drop the TCP, but that
+ # didn't happen in time, this gets True
+ self.wasServerConnectionDropTimeout = False
+
+ # When the initial WebSocket opening handshake times out, this gets True
+ self.wasOpenHandshakeTimeout = False
+
+ # When we initiated a closing handshake, but the peer did not respond in
+ # time, this gets True
+ self.wasCloseHandshakeTimeout = False
+
+ # The close code I sent in close frame (if any)
+ self.localCloseCode = None
+
+ # The close reason I sent in close frame (if any)
+ self.localCloseReason = None
+
+ # The close code the peer sent me in close frame (if any)
+ self.remoteCloseCode = None
+
+ # The close reason the peer sent me in close frame (if any)
+ self.remoteCloseReason = None
+
+ # timers, which might get set up later, and remembered here to get canceled
+ # when appropriate
+ if not self.isServer:
+ self.serverConnectionDropTimeoutCall = None
+ self.openHandshakeTimeoutCall = None
+ self.closeHandshakeTimeoutCall = None
+
+ # set opening handshake timeout handler
+ if self.openHandshakeTimeout > 0:
+ self.openHandshakeTimeoutCall = reactor.callLater(self.openHandshakeTimeout, self.onOpenHandshakeTimeout)
+
+
+ def connectionLost(self, reason):
+ """
+ This is called by Twisted framework when a TCP connection was lost.
+
+ Modes: Hybi, Hixie
+ """
+ ## cancel any server connection drop timer if present
+ ##
+ if not self.isServer and self.serverConnectionDropTimeoutCall is not None:
+ if self.debugCodePaths:
+ log.msg("serverConnectionDropTimeoutCall.cancel")
+ self.serverConnectionDropTimeoutCall.cancel()
+ self.serverConnectionDropTimeoutCall = None
+
+ self.state = WebSocketProtocol.STATE_CLOSED
+ if not self.wasClean:
+ if not self.droppedByMe and self.wasNotCleanReason is None:
+ self.wasNotCleanReason = "peer dropped the TCP connection without previous WebSocket closing handshake"
+ self.onClose(self.wasClean, WebSocketProtocol.CLOSE_STATUS_CODE_ABNORMAL_CLOSE, "connection was closed uncleanly (%s)" % self.wasNotCleanReason)
+ else:
+ self.onClose(self.wasClean, self.remoteCloseCode, self.remoteCloseReason)
+
+
+ def logRxOctets(self, data):
+ """
+ Hook fired right after raw octets have been received, but only when self.logOctets == True.
+
+ Modes: Hybi, Hixie
+ """
+ log.msg("RX Octets from %s : octets = %s" % (self.peerstr, binascii.b2a_hex(data)))
+
+
+ def logTxOctets(self, data, sync):
+ """
+ Hook fired right after raw octets have been sent, but only when self.logOctets == True.
+
+ Modes: Hybi, Hixie
+ """
+ log.msg("TX Octets to %s : sync = %s, octets = %s" % (self.peerstr, sync, binascii.b2a_hex(data)))
+
+
+ def logRxFrame(self, frameHeader, payload):
+ """
+ Hook fired right after WebSocket frame has been received and decoded, but only when self.logFrames == True.
+
+ Modes: Hybi
+ """
+ data = ''.join(payload)
+ info = (self.peerstr,
+ frameHeader.fin,
+ frameHeader.rsv,
+ frameHeader.opcode,
+ binascii.b2a_hex(frameHeader.mask) if frameHeader.mask else "-",
+ frameHeader.length,
+ data if frameHeader.opcode == 1 else binascii.b2a_hex(data))
+
+ log.msg("RX Frame from %s : fin = %s, rsv = %s, opcode = %s, mask = %s, length = %s, payload = %s" % info)
+
+
+ def logTxFrame(self, frameHeader, payload, repeatLength, chopsize, sync):
+ """
+ Hook fired right after WebSocket frame has been encoded and sent, but only when self.logFrames == True.
+
+ Modes: Hybi
+ """
+ info = (self.peerstr,
+ frameHeader.fin,
+ frameHeader.rsv,
+ frameHeader.opcode,
+ binascii.b2a_hex(frameHeader.mask) if frameHeader.mask else "-",
+ frameHeader.length,
+ repeatLength,
+ chopsize,
+ sync,
+ payload if frameHeader.opcode == 1 else binascii.b2a_hex(payload))
+
+ log.msg("TX Frame to %s : fin = %s, rsv = %s, opcode = %s, mask = %s, length = %s, repeat_length = %s, chopsize = %s, sync = %s, payload = %s" % info)
+
+
+ def dataReceived(self, data):
+ """
+ This is called by Twisted framework upon receiving data on TCP connection.
+
+ Modes: Hybi, Hixie
+ """
+ if self.logOctets:
+ self.logRxOctets(data)
+ self.data += data
+ self.consumeData()
+
+
+ def consumeData(self):
+ """
+ Consume buffered (incoming) data.
+
+ Modes: Hybi, Hixie
+ """
+
+ ## WebSocket is open (handshake was completed) or close was sent
+ ##
+ if self.state == WebSocketProtocol.STATE_OPEN or self.state == WebSocketProtocol.STATE_CLOSING:
+
+ ## process until no more buffered data left or WS was closed
+ ##
+ while self.processData() and self.state != WebSocketProtocol.STATE_CLOSED:
+ pass
+
+ ## WebSocket needs handshake
+ ##
+ elif self.state == WebSocketProtocol.STATE_CONNECTING:
+
+ ## the implementation of processHandshake() in derived
+ ## class needs to perform client or server handshake
+ ## from other party here ..
+ ##
+ self.processHandshake()
+
+ ## we failed the connection .. don't process any more data!
+ ##
+ elif self.state == WebSocketProtocol.STATE_CLOSED:
+
+ ## ignore any data received after WS was closed
+ ##
+ if self.debugCodePaths:
+ log.msg("received data in STATE_CLOSED")
+
+ ## should not arrive here (invalid state)
+ ##
+ else:
+ raise Exception("invalid state")
+
+
+ def processHandshake(self):
+ """
+ Process WebSockets handshake.
+
+ Modes: Hybi, Hixie
+ """
+ raise Exception("must implement handshake (client or server) in derived class")
+
+
+ def registerProducer(self, producer, streaming):
+ """
+ Register a Twisted producer with this protocol.
+
+ Modes: Hybi, Hixie
+
+ :param producer: A Twisted push or pull producer.
+ :type producer: object
+ :param streaming: Producer type.
+ :type streaming: bool
+ """
+ self.transport.registerProducer(producer, streaming)
+
+
+ def _trigger(self):
+ """
+ Trigger sending stuff from send queue (which is only used for chopped/synched writes).
+
+ Modes: Hybi, Hixie
+ """
+ if not self.triggered:
+ self.triggered = True
+ self._send()
+
+
+ def _send(self):
+ """
+ Send out stuff from send queue. For details how this works, see test/trickling
+ in the repo.
+
+ Modes: Hybi, Hixie
+ """
+ if len(self.send_queue) > 0:
+ e = self.send_queue.popleft()
+ if self.state != WebSocketProtocol.STATE_CLOSED:
+ self.transport.write(e[0])
+ if self.logOctets:
+ self.logTxOctets(e[0], e[1])
+ else:
+ if self.debugCodePaths:
+ log.msg("skipped delayed write, since connection is closed")
+ # we need to reenter the reactor to make the latter
+ # reenter the OS network stack, so that octets
+ # can get on the wire. Note: this is a "heuristic",
+ # since there is no (easy) way to really force out
+ # octets from the OS network stack to wire.
+ reactor.callLater(WebSocketProtocol.QUEUED_WRITE_DELAY, self._send)
+ else:
+ self.triggered = False
+
+
+ def sendData(self, data, sync = False, chopsize = None):
+ """
+ Wrapper for self.transport.write which allows to give a chopsize.
+ When asked to chop up writing to TCP stream, we write only chopsize octets
+ and then give up control to select() in underlying reactor so that bytes
+ get onto wire immediately. Note that this is different from and unrelated
+ to WebSockets data message fragmentation. Note that this is also different
+ from the TcpNoDelay option which can be set on the socket.
+
+ Modes: Hybi, Hixie
+ """
+ if chopsize and chopsize > 0:
+ i = 0
+ n = len(data)
+ done = False
+ while not done:
+ j = i + chopsize
+ if j >= n:
+ done = True
+ j = n
+ self.send_queue.append((data[i:j], True))
+ i += chopsize
+ self._trigger()
+ else:
+ if sync or len(self.send_queue) > 0:
+ self.send_queue.append((data, sync))
+ self._trigger()
+ else:
+ self.transport.write(data)
+ if self.logOctets:
+ self.logTxOctets(data, False)
+
+
+ def sendPreparedMessage(self, preparedMsg):
+ """
+ Send a message that was previously prepared with
+ WebSocketFactory.prepareMessage().
+
+ Modes: Hybi, Hixie
+ """
+ if self.websocket_version == 0:
+ self.sendData(preparedMsg.payloadHixie)
+ else:
+ self.sendData(preparedMsg.payloadHybi)
+
+
+ def processData(self):
+ """
+ After WebSockets handshake has been completed, this procedure will do all
+ subsequent processing of incoming bytes.
+
+ Modes: Hybi, Hixie
+ """
+ if self.websocket_version == 0:
+ return self.processDataHixie76()
+ else:
+ return self.processDataHybi()
+
+
+ def processDataHixie76(self):
+ """
+ Hixie-76 incoming data processing.
+
+ Modes: Hixie
+ """
+ buffered_len = len(self.data)
+
+ ## outside a message, that is we are awaiting data which starts a new message
+ ##
+ if not self.inside_message:
+ if buffered_len >= 2:
+
+ ## new message
+ ##
+ if self.data[0] == '\x00':
+
+ self.inside_message = True
+
+ if self.utf8validateIncoming:
+ self.utf8validator.reset()
+ self.utf8validateIncomingCurrentMessage = True
+ self.utf8validateLast = (True, True, 0, 0)
+ else:
+ self.utf8validateIncomingCurrentMessage = False
+
+ self.data = self.data[1:]
+ self.onMessageBegin(1)
+
+ ## Hixie close from peer received
+ ##
+ elif self.data[0] == '\xff' and self.data[1] == '\x00':
+ self.onCloseFrame()
+ self.data = self.data[2:]
+ # stop receiving/processing after having received close!
+ return False
+
+ ## malformed data
+ ##
+ else:
+ if self.protocolViolation("malformed data received"):
+ return False
+ else:
+ ## need more data
+ return False
+
+ end_index = self.data.find('\xff')
+ if end_index > 0:
+ payload = self.data[:end_index]
+ self.data = self.data[end_index + 1:]
+ else:
+ payload = self.data
+ self.data = ''
+
+ ## incrementally validate UTF-8 payload
+ ##
+ if self.utf8validateIncomingCurrentMessage:
+ self.utf8validateLast = self.utf8validator.validate(payload)
+ if not self.utf8validateLast[0]:
+ if self.invalidPayload("encountered invalid UTF-8 while processing text message at payload octet index %d" % self.utf8validateLast[3]):
+ return False
+
+ self.onMessageFrameData(payload)
+
+ if end_index > 0:
+ self.inside_message = False
+ self.onMessageEnd()
+
+ return len(self.data) > 0
+
+
+ def processDataHybi(self):
+ """
+ RFC6455/Hybi-Drafts incoming data processing.
+
+ Modes: Hybi
+ """
+ buffered_len = len(self.data)
+
+ ## outside a frame, that is we are awaiting data which starts a new frame
+ ##
+ if self.current_frame is None:
+
+ ## need minimum of 2 octets to for new frame
+ ##
+ if buffered_len >= 2:
+
+ ## FIN, RSV, OPCODE
+ ##
+ b = ord(self.data[0])
+ frame_fin = (b & 0x80) != 0
+ frame_rsv = (b & 0x70) >> 4
+ frame_opcode = b & 0x0f
+
+ ## MASK, PAYLOAD LEN 1
+ ##
+ b = ord(self.data[1])
+ frame_masked = (b & 0x80) != 0
+ frame_payload_len1 = b & 0x7f
+
+ ## MUST be 0 when no extension defining
+ ## the semantics of RSV has been negotiated
+ ##
+ if frame_rsv != 0:
+ if self.protocolViolation("RSV != 0 and no extension negotiated"):
+ return False
+
+ ## all client-to-server frames MUST be masked
+ ##
+ if self.isServer and self.requireMaskedClientFrames and not frame_masked:
+ if self.protocolViolation("unmasked client-to-server frame"):
+ return False
+
+ ## all server-to-client frames MUST NOT be masked
+ ##
+ if not self.isServer and not self.acceptMaskedServerFrames and frame_masked:
+ if self.protocolViolation("masked server-to-client frame"):
+ return False
+
+ ## check frame
+ ##
+ if frame_opcode > 7: # control frame (have MSB in opcode set)
+
+ ## control frames MUST NOT be fragmented
+ ##
+ if not frame_fin:
+ if self.protocolViolation("fragmented control frame"):
+ return False
+
+ ## control frames MUST have payload 125 octets or less
+ ##
+ if frame_payload_len1 > 125:
+ if self.protocolViolation("control frame with payload length > 125 octets"):
+ return False
+
+ ## check for reserved control frame opcodes
+ ##
+ if frame_opcode not in [8, 9, 10]:
+ if self.protocolViolation("control frame using reserved opcode %d" % frame_opcode):
+ return False
+
+ ## close frame : if there is a body, the first two bytes of the body MUST be a 2-byte
+ ## unsigned integer (in network byte order) representing a status code
+ ##
+ if frame_opcode == 8 and frame_payload_len1 == 1:
+ if self.protocolViolation("received close control frame with payload len 1"):
+ return False
+
+ else: # data frame
+
+ ## check for reserved data frame opcodes
+ ##
+ if frame_opcode not in [0, 1, 2]:
+ if self.protocolViolation("data frame using reserved opcode %d" % frame_opcode):
+ return False
+
+ ## check opcode vs message fragmentation state 1/2
+ ##
+ if not self.inside_message and frame_opcode == 0:
+ if self.protocolViolation("received continuation data frame outside fragmented message"):
+ return False
+
+ ## check opcode vs message fragmentation state 2/2
+ ##
+ if self.inside_message and frame_opcode != 0:
+ if self.protocolViolation("received non-continuation data frame while inside fragmented message"):
+ return False
+
+ ## compute complete header length
+ ##
+ if frame_masked:
+ mask_len = 4
+ else:
+ mask_len = 0
+
+ if frame_payload_len1 < 126:
+ frame_header_len = 2 + mask_len
+ elif frame_payload_len1 == 126:
+ frame_header_len = 2 + 2 + mask_len
+ elif frame_payload_len1 == 127:
+ frame_header_len = 2 + 8 + mask_len
+ else:
+ raise Exception("logic error")
+
+ ## only proceed when we have enough data buffered for complete
+ ## frame header (which includes extended payload len + mask)
+ ##
+ if buffered_len >= frame_header_len:
+
+ ## minimum frame header length (already consumed)
+ ##
+ i = 2
+
+ ## extract extended payload length
+ ##
+ if frame_payload_len1 == 126:
+ frame_payload_len = struct.unpack("!H", self.data[i:i+2])[0]
+ if frame_payload_len < 126:
+ if self.protocolViolation("invalid data frame length (not using minimal length encoding)"):
+ return False
+ i += 2
+ elif frame_payload_len1 == 127:
+ frame_payload_len = struct.unpack("!Q", self.data[i:i+8])[0]
+ if frame_payload_len > 0x7FFFFFFFFFFFFFFF: # 2**63
+ if self.protocolViolation("invalid data frame length (>2^63)"):
+ return False
+ if frame_payload_len < 65536:
+ if self.protocolViolation("invalid data frame length (not using minimal length encoding)"):
+ return False
+ i += 8
+ else:
+ frame_payload_len = frame_payload_len1
+
+ ## when payload is masked, extract frame mask
+ ##
+ frame_mask = None
+ if frame_masked:
+ frame_mask = self.data[i:i+4]
+ i += 4
+
+ if frame_masked and frame_payload_len > 0 and self.applyMask:
+ if frame_payload_len < WebSocketProtocol.PAYLOAD_LEN_XOR_BREAKEVEN:
+ self.current_frame_masker = XorMaskerSimple(frame_mask)
+ else:
+ self.current_frame_masker = XorMaskerShifted1(frame_mask)
+ else:
+ self.current_frame_masker = XorMaskerNull()
+
+
+ ## remember rest (payload of current frame after header and everything thereafter)
+ ##
+ self.data = self.data[i:]
+
+ ## ok, got complete frame header
+ ##
+ self.current_frame = FrameHeader(frame_opcode,
+ frame_fin,
+ frame_rsv,
+ frame_payload_len,
+ frame_mask)
+
+ ## process begin on new frame
+ ##
+ self.onFrameBegin()
+
+ ## reprocess when frame has no payload or and buffered data left
+ ##
+ return frame_payload_len == 0 or len(self.data) > 0
+
+ else:
+ return False # need more data
+ else:
+ return False # need more data
+
+ ## inside a started frame
+ ##
+ else:
+
+ ## cut out rest of frame payload
+ ##
+ rest = self.current_frame.length - self.current_frame_masker.pointer()
+ if buffered_len >= rest:
+ data = self.data[:rest]
+ length = rest
+ self.data = self.data[rest:]
+ else:
+ data = self.data
+ length = buffered_len
+ self.data = ""
+
+ if length > 0:
+ ## unmask payload
+ ##
+ payload = self.current_frame_masker.process(data)
+
+ ## process frame data
+ ##
+ fr = self.onFrameData(payload)
+ if fr == False:
+ return False
+
+ ## fire frame end handler when frame payload is complete
+ ##
+ if self.current_frame_masker.pointer() == self.current_frame.length:
+ fr = self.onFrameEnd()
+ if fr == False:
+ return False
+
+ ## reprocess when no error occurred and buffered data left
+ ##
+ return len(self.data) > 0
+
+
+ def onFrameBegin(self):
+ """
+ Begin of receive new frame.
+
+ Modes: Hybi
+ """
+ if self.current_frame.opcode > 7:
+ self.control_frame_data = []
+ else:
+ ## new message started
+ ##
+ if not self.inside_message:
+
+ self.inside_message = True
+
+ if self.current_frame.opcode == WebSocketProtocol.MESSAGE_TYPE_TEXT and self.utf8validateIncoming:
+ self.utf8validator.reset()
+ self.utf8validateIncomingCurrentMessage = True
+ self.utf8validateLast = (True, True, 0, 0)
+ else:
+ self.utf8validateIncomingCurrentMessage = False
+
+ self.onMessageBegin(self.current_frame.opcode)
+
+ self.onMessageFrameBegin(self.current_frame.length, self.current_frame.rsv)
+
+
+ def onFrameData(self, payload):
+ """
+ New data received within frame.
+
+ Modes: Hybi
+ """
+ if self.current_frame.opcode > 7:
+ self.control_frame_data.append(payload)
+ else:
+ ## incrementally validate UTF-8 payload
+ ##
+ if self.utf8validateIncomingCurrentMessage:
+ self.utf8validateLast = self.utf8validator.validate(payload)
+ if not self.utf8validateLast[0]:
+ if self.invalidPayload("encountered invalid UTF-8 while processing text message at payload octet index %d" % self.utf8validateLast[3]):
+ return False
+
+ self.onMessageFrameData(payload)
+
+
+ def onFrameEnd(self):
+ """
+ End of frame received.
+
+ Modes: Hybi
+ """
+ if self.current_frame.opcode > 7:
+ if self.logFrames:
+ self.logRxFrame(self.current_frame, self.control_frame_data)
+ self.processControlFrame()
+ else:
+ if self.logFrames:
+ self.logRxFrame(self.current_frame, self.frame_data)
+ self.onMessageFrameEnd()
+ if self.current_frame.fin:
+ if self.utf8validateIncomingCurrentMessage:
+ if not self.utf8validateLast[1]:
+ if self.invalidPayload("UTF-8 text message payload ended within Unicode code point at payload octet index %d" % self.utf8validateLast[3]):
+ return False
+ self.onMessageEnd()
+ self.inside_message = False
+ self.current_frame = None
+
+
+ def processControlFrame(self):
+ """
+ Process a completely received control frame.
+
+ Modes: Hybi
+ """
+
+ payload = ''.join(self.control_frame_data)
+ self.control_frame_data = None
+
+ ## CLOSE frame
+ ##
+ if self.current_frame.opcode == 8:
+
+ code = None
+ reasonRaw = None
+ ll = len(payload)
+ if ll > 1:
+ code = struct.unpack("!H", payload[0:2])[0]
+ if ll > 2:
+ reasonRaw = payload[2:]
+
+ if self.onCloseFrame(code, reasonRaw):
+ return False
+
+ ## PING frame
+ ##
+ elif self.current_frame.opcode == 9:
+ self.onPing(payload)
+
+ ## PONG frame
+ ##
+ elif self.current_frame.opcode == 10:
+ self.onPong(payload)
+
+ else:
+ ## we might arrive here, when protocolViolation
+ ## wants us to continue anyway
+ pass
+
+ return True
+
+
+ def sendFrame(self, opcode, payload = "", fin = True, rsv = 0, mask = None, payload_len = None, chopsize = None, sync = False):
+ """
+ Send out frame. Normally only used internally via sendMessage(), sendPing(), sendPong() and sendClose().
+
+ This method deliberately allows to send invalid frames (that is frames invalid
+ per-se, or frames invalid because of protocol state). Other than in fuzzing servers,
+ calling methods will ensure that no invalid frames are sent.
+
+ In addition, this method supports explicit specification of payload length.
+ When payload_len is given, it will always write that many octets to the stream.
+ It'll wrap within payload, resending parts of that when more octets were requested
+ The use case is again for fuzzing server which want to sent increasing amounts
+ of payload data to peers without having to construct potentially large messges
+ themselfes.
+
+ Modes: Hybi
+ """
+ if self.websocket_version == 0:
+ raise Exception("function not supported in Hixie-76 mode")
+
+ if payload_len is not None:
+ if len(payload) < 1:
+ raise Exception("cannot construct repeated payload with length %d from payload of length %d" % (payload_len, len(payload)))
+ l = payload_len
+ pl = ''.join([payload for k in range(payload_len / len(payload))]) + payload[:payload_len % len(payload)]
+ else:
+ l = len(payload)
+ pl = payload
+
+ ## first byte
+ ##
+ b0 = 0
+ if fin:
+ b0 |= (1 << 7)
+ b0 |= (rsv % 8) << 4
+ b0 |= opcode % 128
+
+ ## second byte, payload len bytes and mask
+ ##
+ b1 = 0
+ if mask or (not self.isServer and self.maskClientFrames) or (self.isServer and self.maskServerFrames):
+ b1 |= 1 << 7
+ if not mask:
+ mask = struct.pack("!I", random.getrandbits(32))
+ mv = mask
+ else:
+ mv = ""
+
+ ## mask frame payload
+ ##
+ if l > 0 and self.applyMask:
+ if l < WebSocketProtocol.PAYLOAD_LEN_XOR_BREAKEVEN:
+ masker = XorMaskerSimple(mask)
+ else:
+ masker = XorMaskerShifted1(mask)
+ plm = masker.process(pl)
+ else:
+ plm = pl
+
+ else:
+ mv = ""
+ plm = pl
+
+ el = ""
+ if l <= 125:
+ b1 |= l
+ elif l <= 0xFFFF:
+ b1 |= 126
+ el = struct.pack("!H", l)
+ elif l <= 0x7FFFFFFFFFFFFFFF:
+ b1 |= 127
+ el = struct.pack("!Q", l)
+ else:
+ raise Exception("invalid payload length")
+
+ raw = ''.join([chr(b0), chr(b1), el, mv, plm])
+
+ if self.logFrames:
+ frameHeader = FrameHeader(opcode, fin, rsv, l, mask)
+ self.logTxFrame(frameHeader, payload, payload_len, chopsize, sync)
+
+ ## send frame octets
+ ##
+ self.sendData(raw, sync, chopsize)
+
+
+ def sendPing(self, payload = None):
+ """
+ Send out Ping to peer. A peer is expected to Pong back the payload a soon
+ as "practical". When more than 1 Ping is outstanding at a peer, the peer may
+ elect to respond only to the last Ping.
+
+ Modes: Hybi
+
+ :param payload: An optional, arbitrary payload of length < 126 octets.
+ :type payload: str
+ """
+ if self.websocket_version == 0:
+ raise Exception("function not supported in Hixie-76 mode")
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+ if payload:
+ l = len(payload)
+ if l > 125:
+ raise Exception("invalid payload for PING (payload length must be <= 125, was %d)" % l)
+ self.sendFrame(opcode = 9, payload = payload)
+ else:
+ self.sendFrame(opcode = 9)
+
+
+ def sendPong(self, payload = None):
+ """
+ Send out Pong to peer. A Pong frame MAY be sent unsolicited.
+ This serves as a unidirectional heartbeat. A response to an unsolicited pong is "not expected".
+
+ Modes: Hybi
+
+ :param payload: An optional, arbitrary payload of length < 126 octets.
+ :type payload: str
+ """
+ if self.websocket_version == 0:
+ raise Exception("function not supported in Hixie-76 mode")
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+ if payload:
+ l = len(payload)
+ if l > 125:
+ raise Exception("invalid payload for PONG (payload length must be <= 125, was %d)" % l)
+ self.sendFrame(opcode = 10, payload = payload)
+ else:
+ self.sendFrame(opcode = 10)
+
+
+ def sendCloseFrame(self, code = None, reasonUtf8 = None, isReply = False):
+ """
+ Send a close frame and update protocol state. Note, that this is
+ an internal method which deliberately allows not send close
+ frame with invalid payload.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, this method is slightly misnamed for historic reasons.
+ - For Hixie mode, code and reasonUtf8 will be silently ignored.
+ """
+ if self.state == WebSocketProtocol.STATE_CLOSING:
+ if self.debugCodePaths:
+ log.msg("ignoring sendCloseFrame since connection is closing")
+
+ elif self.state == WebSocketProtocol.STATE_CLOSED:
+ if self.debugCodePaths:
+ log.msg("ignoring sendCloseFrame since connection already closed")
+
+ elif self.state == WebSocketProtocol.STATE_CONNECTING:
+ raise Exception("cannot close a connection not yet connected")
+
+ elif self.state == WebSocketProtocol.STATE_OPEN:
+
+ if self.websocket_version == 0:
+ self.sendData("\xff\x00")
+ else:
+ ## construct Hybi close frame payload and send frame
+ payload = ""
+ if code is not None:
+ payload += struct.pack("!H", code)
+ if reasonUtf8 is not None:
+ payload += reasonUtf8
+ self.sendFrame(opcode = 8, payload = payload)
+
+ ## update state
+ self.state = WebSocketProtocol.STATE_CLOSING
+ self.closedByMe = not isReply
+
+ ## remember payload of close frame we sent
+ self.localCloseCode = code
+ self.localCloseReason = reasonUtf8
+
+ ## drop connection when timeout on receiving close handshake reply
+ if self.closedByMe and self.closeHandshakeTimeout > 0:
+ self.closeHandshakeTimeoutCall = reactor.callLater(self.closeHandshakeTimeout, self.onCloseHandshakeTimeout)
+
+ else:
+ raise Exception("logic error")
+
+
+ def sendClose(self, code = None, reason = None):
+ """
+ Starts a closing handshake.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, code and reason will be silently ignored.
+
+ :param code: An optional close status code (:class:`WebSocketProtocol`.CLOSE_STATUS_CODE_NORMAL or 3000-4999).
+ :type code: int
+ :param reason: An optional close reason (a string that when present, a status code MUST also be present).
+ :type reason: str
+ """
+ if code is not None:
+ if type(code) != int:
+ raise Exception("invalid type %s for close code" % type(code))
+ if code != 1000 and not (code >= 3000 and code <= 4999):
+ raise Exception("invalid close code %d" % code)
+ if reason is not None:
+ if code is None:
+ raise Exception("close reason without close code")
+ if type(reason) not in [str, unicode]:
+ raise Exception("invalid type %s for close reason" % type(reason))
+ reasonUtf8 = reason.encode("UTF-8")
+ if len(reasonUtf8) + 2 > 125:
+ raise Exception("close reason too long (%d)" % len(reasonUtf8))
+ else:
+ reasonUtf8 = None
+ self.sendCloseFrame(code = code, reasonUtf8 = reasonUtf8, isReply = False)
+
+
+ def beginMessage(self, opcode = MESSAGE_TYPE_TEXT):
+ """
+ Begin sending new message.
+
+ Modes: Hybi, Hixie
+
+ :param opcode: Message type, normally either WebSocketProtocol.MESSAGE_TYPE_TEXT (default) or
+ WebSocketProtocol.MESSAGE_TYPE_BINARY (only Hybi mode).
+ """
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+
+ ## check if sending state is valid for this method
+ ##
+ if self.send_state != WebSocketProtocol.SEND_STATE_GROUND:
+ raise Exception("WebSocketProtocol.beginMessage invalid in current sending state")
+
+ if self.websocket_version == 0:
+ if opcode != 1:
+ raise Exception("cannot send non-text message in Hixie mode")
+
+ self.sendData('\x00')
+ self.send_state = WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE
+ else:
+ if opcode not in [1, 2]:
+ raise Exception("use of reserved opcode %d" % opcode)
+
+ ## remember opcode for later (when sending first frame)
+ ##
+ self.send_message_opcode = opcode
+ self.send_state = WebSocketProtocol.SEND_STATE_MESSAGE_BEGIN
+
+
+
+ def beginMessageFrame(self, length, reserved = 0, mask = None):
+ """
+ Begin sending new message frame.
+
+ Modes: Hybi
+
+ :param length: Length of frame which is started. Must be >= 0 and <= 2^63.
+ :type length: int
+ :param reserved: Reserved bits for frame (an integer from 0 to 7). Note that reserved != 0 is only legal when an extension has been negoiated which defines semantics.
+ :type reserved: int
+ :param mask: Optional frame mask. When given, this is used. When None and the peer is a client, a mask will be internally generated. For servers None is default.
+ :type mask: str
+ """
+ if self.websocket_version == 0:
+ raise Exception("function not supported in Hixie-76 mode")
+
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+ ## check if sending state is valid for this method
+ ##
+ if self.send_state not in [WebSocketProtocol.SEND_STATE_MESSAGE_BEGIN, WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE]:
+ raise Exception("WebSocketProtocol.beginMessageFrame invalid in current sending state")
+
+ if (not type(length) in [int, long]) or length < 0 or length > 0x7FFFFFFFFFFFFFFF: # 2**63
+ raise Exception("invalid value for message frame length")
+
+ if type(reserved) is not int or reserved < 0 or reserved > 7:
+ raise Exception("invalid value for reserved bits")
+
+ self.send_message_frame_length = length
+
+ if mask:
+ ## explicit mask given
+ ##
+ assert type(mask) == str
+ assert len(mask) == 4
+ self.send_message_frame_mask = mask
+
+ elif (not self.isServer and self.maskClientFrames) or (self.isServer and self.maskServerFrames):
+ ## automatic mask:
+ ## - client-to-server masking (if not deactivated)
+ ## - server-to-client masking (if activated)
+ ##
+ self.send_message_frame_mask = struct.pack("!I", random.getrandbits(32))
+
+ else:
+ ## no mask
+ ##
+ self.send_message_frame_mask = None
+
+ ## payload masker
+ ##
+ if self.send_message_frame_mask and length > 0 and self.applyMask:
+ if length < WebSocketProtocol.PAYLOAD_LEN_XOR_BREAKEVEN:
+ self.send_message_frame_masker = XorMaskerSimple(self.send_message_frame_mask)
+ else:
+ self.send_message_frame_masker = XorMaskerShifted1(self.send_message_frame_mask)
+ else:
+ self.send_message_frame_masker = XorMaskerNull()
+
+ ## first byte
+ ##
+ b0 = (reserved % 8) << 4 # FIN = false .. since with streaming, we don't know when message ends
+
+ if self.send_state == WebSocketProtocol.SEND_STATE_MESSAGE_BEGIN:
+ self.send_state = WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE
+ b0 |= self.send_message_opcode % 128
+ else:
+ pass # message continuation frame
+
+ ## second byte, payload len bytes and mask
+ ##
+ b1 = 0
+ if self.send_message_frame_mask:
+ b1 |= 1 << 7
+ mv = self.send_message_frame_mask
+ else:
+ mv = ""
+
+ el = ""
+ if length <= 125:
+ b1 |= length
+ elif length <= 0xFFFF:
+ b1 |= 126
+ el = struct.pack("!H", length)
+ elif length <= 0x7FFFFFFFFFFFFFFF:
+ b1 |= 127
+ el = struct.pack("!Q", length)
+ else:
+ raise Exception("invalid payload length")
+
+ ## write message frame header
+ ##
+ header = ''.join([chr(b0), chr(b1), el, mv])
+ self.sendData(header)
+
+ ## now we are inside message frame ..
+ ##
+ self.send_state = WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE_FRAME
+
+
+ def sendMessageFrameData(self, payload, sync = False):
+ """
+ Send out data when within message frame (message was begun, frame was begun).
+ Note that the frame is automatically ended when enough data has been sent
+ that is, there is no endMessageFrame, since you have begun the frame specifying
+ the frame length, which implicitly defined the frame end. This is different from
+ messages, which you begin and end, since a message can contain an unlimited number
+ of frames.
+
+ Modes: Hybi, Hixie
+
+ Notes:
+ - For Hixie mode, this method is slightly misnamed for historic reasons.
+
+ :param payload: Data to send.
+
+ :returns: int -- Hybi mode: when frame still incomplete, returns outstanding octets, when frame complete, returns <= 0, when < 0, the amount of unconsumed data in payload argument. Hixie mode: returns None.
+ """
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+
+ if self.websocket_version == 0:
+ ## Hixie Mode
+ ##
+ if self.send_state != WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE:
+ raise Exception("WebSocketProtocol.sendMessageFrameData invalid in current sending state")
+ self.sendData(payload, sync = sync)
+ return None
+
+ else:
+ ## Hybi Mode
+ ##
+ if self.send_state != WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE_FRAME:
+ raise Exception("WebSocketProtocol.sendMessageFrameData invalid in current sending state")
+
+ rl = len(payload)
+ if self.send_message_frame_masker.pointer() + rl > self.send_message_frame_length:
+ l = self.send_message_frame_length - self.send_message_frame_masker.pointer()
+ rest = -(rl - l)
+ pl = payload[:l]
+ else:
+ l = rl
+ rest = self.send_message_frame_length - self.send_message_frame_masker.pointer() - l
+ pl = payload
+
+ ## mask frame payload
+ ##
+ plm = self.send_message_frame_masker.process(pl)
+
+ ## send frame payload
+ ##
+ self.sendData(plm, sync = sync)
+
+ ## if we are done with frame, move back into "inside message" state
+ ##
+ if self.send_message_frame_masker.pointer() >= self.send_message_frame_length:
+ self.send_state = WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE
+
+ ## when =0 : frame was completed exactly
+ ## when >0 : frame is still uncomplete and that much amount is still left to complete the frame
+ ## when <0 : frame was completed and there was this much unconsumed data in payload argument
+ ##
+ return rest
+
+
+ def endMessage(self):
+ """
+ End a previously begun message. No more frames may be sent (for that message). You have to
+ begin a new message before sending again.
+
+ Modes: Hybi, Hixie
+ """
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+ ## check if sending state is valid for this method
+ ##
+ if self.send_state != WebSocketProtocol.SEND_STATE_INSIDE_MESSAGE:
+ raise Exception("WebSocketProtocol.endMessage invalid in current sending state [%d]" % self.send_state)
+
+ if self.websocket_version == 0:
+ self.sendData('\x00')
+ else:
+ self.sendFrame(opcode = 0, fin = True)
+
+ self.send_state = WebSocketProtocol.SEND_STATE_GROUND
+
+
+ def sendMessageFrame(self, payload, reserved = 0, mask = None, sync = False):
+ """
+ When a message has begun, send a complete message frame in one go.
+
+ Modes: Hybi
+ """
+ if self.websocket_version == 0:
+ raise Exception("function not supported in Hixie-76 mode")
+
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+ if self.websocket_version == 0:
+ raise Exception("function not supported in Hixie-76 mode")
+ self.beginMessageFrame(len(payload), reserved, mask)
+ self.sendMessageFrameData(payload, sync)
+
+
+ def sendMessage(self, payload, binary = False, payload_frag_size = None, sync = False):
+ """
+ Send out a message in one go.
+
+ You can send text or binary message, and optionally specifiy a payload fragment size.
+ When the latter is given, the payload will be split up into frames with
+ payload <= the payload_frag_size given.
+
+ Modes: Hybi, Hixie
+ """
+ if self.state != WebSocketProtocol.STATE_OPEN:
+ return
+ if self.websocket_version == 0:
+ if binary:
+ raise Exception("cannot send binary message in Hixie76 mode")
+ if payload_frag_size:
+ raise Exception("cannot fragment messages in Hixie76 mode")
+ self.sendMessageHixie76(payload, sync)
+ else:
+ self.sendMessageHybi(payload, binary, payload_frag_size, sync)
+
+
+ def sendMessageHixie76(self, payload, sync = False):
+ """
+ Hixie76-Variant of sendMessage().
+
+ Modes: Hixie
+ """
+ self.sendData('\x00' + payload + '\xff', sync = sync)
+
+
+ def sendMessageHybi(self, payload, binary = False, payload_frag_size = None, sync = False):
+ """
+ Hybi-Variant of sendMessage().
+
+ Modes: Hybi
+ """
+ ## (initial) frame opcode
+ ##
+ if binary:
+ opcode = 2
+ else:
+ opcode = 1
+
+ ## explicit payload_frag_size arguments overrides autoFragmentSize setting
+ ##
+ if payload_frag_size is not None:
+ pfs = payload_frag_size
+ else:
+ if self.autoFragmentSize > 0:
+ pfs = self.autoFragmentSize
+ else:
+ pfs = None
+
+ ## send unfragmented
+ ##
+ if pfs is None or len(payload) <= pfs:
+ self.sendFrame(opcode = opcode, payload = payload, sync = sync)
+
+ ## send data message in fragments
+ ##
+ else:
+ if pfs < 1:
+ raise Exception("payload fragment size must be at least 1 (was %d)" % pfs)
+ n = len(payload)
+ i = 0
+ done = False
+ first = True
+ while not done:
+ j = i + pfs
+ if j > n:
+ done = True
+ j = n
+ if first:
+ self.sendFrame(opcode = opcode, payload = payload[i:j], fin = done, sync = sync)
+ first = False
+ else:
+ self.sendFrame(opcode = 0, payload = payload[i:j], fin = done, sync = sync)
+ i += pfs
+
+
+
+class PreparedMessage:
+ """
+ Encapsulates a prepared message to be sent later once or multiple
+ times. This is used for optimizing Broadcast/PubSub.
+
+ The message serialization formats currently created internally are:
+ * Hybi
+ * Hixie
+
+ The construction of different formats is needed, since we support
+ mixed clients (speaking different protocol versions).
+
+ It will also be the place to add a 3rd format, when we support
+ the deflate extension, since then, the clients will be mixed
+ between Hybi-Deflate-Unsupported, Hybi-Deflate-Supported and Hixie.
+ """
+
+ def __init__(self, payload, binary, masked):
+ self.initHixie(payload, binary)
+ self.initHybi(payload, binary, masked)
+
+
+ def initHixie(self, payload, binary):
+ if binary:
+ # silently filter out .. probably do something else:
+ # base64?
+ # dunno
+ self.payloadHixie = ''
+ else:
+ self.payloadHixie = '\x00' + payload + '\xff'
+
+
+ def initHybi(self, payload, binary, masked):
+ l = len(payload)
+
+ ## first byte
+ ##
+ b0 = ((1 << 7) | 2) if binary else ((1 << 7) | 1)
+
+ ## second byte, payload len bytes and mask
+ ##
+ if masked:
+ b1 = 1 << 7
+ mask = struct.pack("!I", random.getrandbits(32))
+ if l == 0:
+ plm = payload
+ elif l < WebSocketProtocol.PAYLOAD_LEN_XOR_BREAKEVEN:
+ plm = XorMaskerSimple(mask).process(payload)
+ else:
+ plm = XorMaskerShifted1(mask).process(payload)
+ else:
+ b1 = 0
+ mask = ""
+ plm = payload
+
+ ## payload extended length
+ ##
+ el = ""
+ if l <= 125:
+ b1 |= l
+ elif l <= 0xFFFF:
+ b1 |= 126
+ el = struct.pack("!H", l)
+ elif l <= 0x7FFFFFFFFFFFFFFF:
+ b1 |= 127
+ el = struct.pack("!Q", l)
+ else:
+ raise Exception("invalid payload length")
+
+ ## raw WS message (single frame)
+ ##
+ self.payloadHybi = ''.join([chr(b0), chr(b1), el, mask, plm])
+
+
+
+class WebSocketFactory:
+ """
+ Mixin for WebSocketClientFactory and WebSocketServerFactory.
+ """
+
+ def prepareMessage(self, payload, binary = False, masked = None):
+ """
+ Prepare a WebSocket message. This can be later used on multiple
+ instances of WebSocketProtocol using sendPreparedMessage().
+
+ By doing so, you can avoid the (small) overhead of framing the
+ _same_ payload into WS messages when that payload is to be sent
+ out on multiple connections.
+
+ Modes: Hybi, Hixie
+
+ Caveats:
+
+ 1) Only use when you know what you are doing. I.e. calling
+ sendPreparedMessage() on the _same_ protocol instance multiples
+ times with the same prepared message might break the spec.
+ Since i.e. the frame mask will be the same!
+
+ 2) Treat the object returned as opaque. It may change!
+ """
+ if masked is None:
+ masked = not self.isServer
+
+ return PreparedMessage(payload, binary, masked)
+
+
+
+class WebSocketServerProtocol(WebSocketProtocol):
+ """
+ A Twisted protocol for WebSockets servers.
+ """
+
+ def onConnect(self, connectionRequest):
+ """
+ Callback fired during WebSocket opening handshake when new WebSocket client
+ connection is about to be established.
+
+ Throw HttpException when you don't want to accept the WebSocket
+ connection request. For example, throw a
+ HttpException(httpstatus.HTTP_STATUS_CODE_UNAUTHORIZED[0], "You are not authorized for this!").
+
+ When you want to accept the connection, return the accepted protocol
+ from list of WebSockets (sub)protocols provided by client or None to
+ speak no specific one or when the client list was empty.
+
+ :param connectionRequest: WebSocket connection request information.
+ :type connectionRequest: instance of :class:`autobahn.websocket.ConnectionRequest`
+ """
+ return None
+
+
+ def connectionMade(self):
+ """
+ Called by Twisted when new TCP connection from client was accepted. Default
+ implementation will prepare for initial WebSocket opening handshake.
+ When overriding in derived class, make sure to call this base class
+ implementation _before_ your code.
+ """
+ self.isServer = True
+ WebSocketProtocol.connectionMade(self)
+ self.factory.countConnections += 1
+ if self.debug:
+ log.msg("connection accepted from peer %s" % self.peerstr)
+
+
+ def connectionLost(self, reason):
+ """
+ Called by Twisted when established TCP connection from client was lost. Default
+ implementation will tear down all state properly.
+ When overriding in derived class, make sure to call this base class
+ implementation _after_ your code.
+ """
+ WebSocketProtocol.connectionLost(self, reason)
+ self.factory.countConnections -= 1
+ if self.debug:
+ log.msg("connection from %s lost" % self.peerstr)
+
+
+ def parseHixie76Key(self, key):
+ return int(filter(lambda x: x.isdigit(), key)) / key.count(" ")
+
+
+ def processHandshake(self):
+ """
+ Process WebSockets opening handshake request from client.
+ """
+ ## only proceed when we have fully received the HTTP request line and all headers
+ ##
+ end_of_header = self.data.find("\x0d\x0a\x0d\x0a")
+ if end_of_header >= 0:
+
+ self.http_request_data = self.data[:end_of_header + 4]
+ if self.debug:
+ log.msg("received HTTP request:\n\n%s\n\n" % self.http_request_data)
+
+ ## extract HTTP status line and headers
+ ##
+ (self.http_status_line, self.http_headers, http_headers_cnt) = parseHttpHeader(self.http_request_data)
+
+ ## validate WebSocket opening handshake client request
+ ##
+ if self.debug:
+ log.msg("received HTTP status line in opening handshake : %s" % str(self.http_status_line))
+ log.msg("received HTTP headers in opening handshake : %s" % str(self.http_headers))
+
+ ## HTTP Request line : METHOD, VERSION
+ ##
+ rl = self.http_status_line.split()
+ if len(rl) != 3:
+ return self.failHandshake("Bad HTTP request status line '%s'" % self.http_status_line)
+ if rl[0].strip() != "GET":
+ return self.failHandshake("HTTP method '%s' not allowed" % rl[0], HTTP_STATUS_CODE_METHOD_NOT_ALLOWED[0])
+ vs = rl[2].strip().split("/")
+ if len(vs) != 2 or vs[0] != "HTTP" or vs[1] not in ["1.1"]:
+ return self.failHandshake("Unsupported HTTP version '%s'" % rl[2], HTTP_STATUS_CODE_UNSUPPORTED_HTTP_VERSION[0])
+
+ ## HTTP Request line : REQUEST-URI
+ ##
+ self.http_request_uri = rl[1].strip()
+ try:
+ (scheme, netloc, path, params, query, fragment) = urlparse.urlparse(self.http_request_uri)
+
+ ## FIXME: check that if absolute resource URI is given,
+ ## the scheme/netloc matches the server
+ if scheme != "" or netloc != "":
+ pass
+
+ ## Fragment identifiers are meaningless in the context of WebSocket
+ ## URIs, and MUST NOT be used on these URIs.
+ if fragment != "":
+ return self.failHandshake("HTTP requested resource contains a fragment identifier '%s'" % fragment)
+
+ ## resource path and query parameters .. this will get forwarded
+ ## to onConnect()
+ self.http_request_path = path
+ self.http_request_params = urlparse.parse_qs(query)
+ except:
+ return self.failHandshake("Bad HTTP request resource - could not parse '%s'" % rl[1].strip())
+
+ ## Host
+ ##
+ if not self.http_headers.has_key("host"):
+ return self.failHandshake("HTTP Host header missing in opening handshake request")
+ if http_headers_cnt["host"] > 1:
+ return self.failHandshake("HTTP Host header appears more than once in opening handshake request")
+ self.http_request_host = self.http_headers["host"].strip()
+ if self.http_request_host.find(":") >= 0:
+ (h, p) = self.http_request_host.split(":")
+ try:
+ port = int(str(p.strip()))
+ except:
+ return self.failHandshake("invalid port '%s' in HTTP Host header '%s'" % (str(p.strip()), str(self.http_request_host)))
+ if port != self.factory.port:
+ return self.failHandshake("port %d in HTTP Host header '%s' does not match server listening port %s" % (port, str(self.http_request_host), self.factory.port))
+ self.http_request_host = h
+ else:
+ if not ((self.factory.isSecure and self.factory.port == 443) or (not self.factory.isSecure and self.factory.port == 80)):
+ return self.failHandshake("missing port in HTTP Host header '%s' and server runs on non-standard port %d (wss = %s)" % (str(self.http_request_host), self.factory.port, self.factory.isSecure))
+
+ ## Upgrade
+ ##
+ if not self.http_headers.has_key("upgrade"):
+ ## When no WS upgrade, render HTML server status page
+ ##
+ if self.webStatus:
+ self.sendServerStatus()
+ self.dropConnection(abort = False)
+ return
+ else:
+ return self.failHandshake("HTTP Upgrade header missing", HTTP_STATUS_CODE_UPGRADE_REQUIRED[0])
+ upgradeWebSocket = False
+ for u in self.http_headers["upgrade"].split(","):
+ if u.strip().lower() == "websocket":
+ upgradeWebSocket = True
+ break
+ if not upgradeWebSocket:
+ return self.failHandshake("HTTP Upgrade headers do not include 'websocket' value (case-insensitive) : %s" % self.http_headers["upgrade"])
+
+ ## Connection
+ ##
+ if not self.http_headers.has_key("connection"):
+ return self.failHandshake("HTTP Connection header missing")
+ connectionUpgrade = False
+ for c in self.http_headers["connection"].split(","):
+ if c.strip().lower() == "upgrade":
+ connectionUpgrade = True
+ break
+ if not connectionUpgrade:
+ return self.failHandshake("HTTP Connection headers do not include 'upgrade' value (case-insensitive) : %s" % self.http_headers["connection"])
+
+ ## Sec-WebSocket-Version PLUS determine mode: Hybi or Hixie
+ ##
+ if not self.http_headers.has_key("sec-websocket-version"):
+ if self.debugCodePaths:
+ log.msg("Hixie76 protocol detected")
+ if self.allowHixie76:
+ version = 0
+ else:
+ return self.failHandshake("WebSocket connection denied - Hixie76 protocol mode disabled.")
+ else:
+ if self.debugCodePaths:
+ log.msg("Hybi protocol detected")
+ if http_headers_cnt["sec-websocket-version"] > 1:
+ return self.failHandshake("HTTP Sec-WebSocket-Version header appears more than once in opening handshake request")
+ try:
+ version = int(self.http_headers["sec-websocket-version"])
+ except:
+ return self.failHandshake("could not parse HTTP Sec-WebSocket-Version header '%s' in opening handshake request" % self.http_headers["sec-websocket-version"])
+
+ if version not in self.versions:
+
+ ## respond with list of supported versions (descending order)
+ ##
+ sv = sorted(self.versions)
+ sv.reverse()
+ svs = ','.join([str(x) for x in sv])
+ return self.failHandshake("WebSocket version %d not supported (supported versions: %s)" % (version, svs),
+ HTTP_STATUS_CODE_BAD_REQUEST[0],
+ [("Sec-WebSocket-Version", svs)])
+ else:
+ ## store the protocol version we are supposed to talk
+ self.websocket_version = version
+
+ ## Sec-WebSocket-Protocol
+ ##
+ if self.http_headers.has_key("sec-websocket-protocol"):
+ protocols = [str(x.strip()) for x in self.http_headers["sec-websocket-protocol"].split(",")]
+ # check for duplicates in protocol header
+ pp = {}
+ for p in protocols:
+ if pp.has_key(p):
+ return self.failHandshake("duplicate protocol '%s' specified in HTTP Sec-WebSocket-Protocol header" % p)
+ else:
+ pp[p] = 1
+ # ok, no duplicates, save list in order the client sent it
+ self.websocket_protocols = protocols
+ else:
+ self.websocket_protocols = []
+
+ ## Origin / Sec-WebSocket-Origin
+ ## http://tools.ietf.org/html/draft-ietf-websec-origin-02
+ ##
+ if self.websocket_version < 13 and self.websocket_version != 0:
+ # Hybi, but only < Hybi-13
+ websocket_origin_header_key = 'sec-websocket-origin'
+ else:
+ # RFC6455, >= Hybi-13 and Hixie
+ websocket_origin_header_key = "origin"
+
+ self.websocket_origin = None
+ if self.http_headers.has_key(websocket_origin_header_key):
+ if http_headers_cnt[websocket_origin_header_key] > 1:
+ return self.failHandshake("HTTP Origin header appears more than once in opening handshake request")
+ self.websocket_origin = self.http_headers[websocket_origin_header_key].strip()
+ else:
+ # non-browser clients are allowed to omit this header
+ pass
+
+ ## Sec-WebSocket-Extensions
+ ##
+ ## extensions requested by client
+ self.websocket_extensions = []
+ ## extensions selected by server
+ self.websocket_extensions_in_use = []
+
+ if self.http_headers.has_key("sec-websocket-extensions"):
+ if self.websocket_version == 0:
+ return self.failHandshake("Sec-WebSocket-Extensions header specified for Hixie-76")
+ extensions = [x.strip() for x in self.http_headers["sec-websocket-extensions"].split(',')]
+ if len(extensions) > 0:
+ self.websocket_extensions = extensions
+ if self.debug:
+ log.msg("client requested extensions we don't support (%s)" % str(extensions))
+
+ ## Sec-WebSocket-Key (Hybi) or Sec-WebSocket-Key1/Sec-WebSocket-Key2 (Hixie-76)
+ ##
+ if self.websocket_version == 0:
+ for kk in ['Sec-WebSocket-Key1', 'Sec-WebSocket-Key2']:
+ k = kk.lower()
+ if not self.http_headers.has_key(k):
+ return self.failHandshake("HTTP %s header missing" % kk)
+ if http_headers_cnt[k] > 1:
+ return self.failHandshake("HTTP %s header appears more than once in opening handshake request" % kk)
+ try:
+ key1 = self.parseHixie76Key(self.http_headers["sec-websocket-key1"].strip())
+ key2 = self.parseHixie76Key(self.http_headers["sec-websocket-key2"].strip())
+ except:
+ return self.failHandshake("could not parse Sec-WebSocket-Key1/2")
+ else:
+ if not self.http_headers.has_key("sec-websocket-key"):
+ return self.failHandshake("HTTP Sec-WebSocket-Key header missing")
+ if http_headers_cnt["sec-websocket-key"] > 1:
+ return self.failHandshake("HTTP Sec-WebSocket-Key header appears more than once in opening handshake request")
+ key = self.http_headers["sec-websocket-key"].strip()
+ if len(key) != 24: # 16 bytes => (ceil(128/24)*24)/6 == 24
+ return self.failHandshake("bad Sec-WebSocket-Key (length must be 24 ASCII chars) '%s'" % key)
+ if key[-2:] != "==": # 24 - ceil(128/6) == 2
+ return self.failHandshake("bad Sec-WebSocket-Key (invalid base64 encoding) '%s'" % key)
+ for c in key[:-2]:
+ if c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/":
+ return self.failHandshake("bad character '%s' in Sec-WebSocket-Key (invalid base64 encoding) '%s'" (c, key))
+
+ ## For Hixie-76, we need 8 octets of HTTP request body to complete HS!
+ ##
+ if self.websocket_version == 0:
+ if len(self.data) < end_of_header + 4 + 8:
+ return
+ else:
+ key3 = self.data[end_of_header + 4:end_of_header + 4 + 8]
+
+ ## Ok, got complete HS input, remember rest (if any)
+ ##
+ if self.websocket_version == 0:
+ self.data = self.data[end_of_header + 4 + 8:]
+ else:
+ self.data = self.data[end_of_header + 4:]
+
+ ## WebSocket handshake validated => produce opening handshake response
+
+ ## Now fire onConnect() on derived class, to give that class a chance to accept or deny
+ ## the connection. onConnect() may throw, in which case the connection is denied, or it
+ ## may return a protocol from the protocols provided by client or None.
+ ##
+ try:
+ connectionRequest = ConnectionRequest(self.peer,
+ self.peerstr,
+ self.http_headers,
+ self.http_request_host,
+ self.http_request_path,
+ self.http_request_params,
+ self.websocket_version,
+ self.websocket_origin,
+ self.websocket_protocols,
+ self.websocket_extensions)
+
+ ## onConnect() will return the selected subprotocol or None
+ ## or raise an HttpException
+ ##
+ protocol = self.onConnect(connectionRequest)
+
+ if protocol is not None and not (protocol in self.websocket_protocols):
+ raise Exception("protocol accepted must be from the list client sent or None")
+
+ self.websocket_protocol_in_use = protocol
+
+ except HttpException, e:
+ return self.failHandshake(e.reason, e.code)
+ #return self.sendHttpRequestFailure(e.code, e.reason)
+
+ except Exception, e:
+ log.msg("Exception raised in onConnect() - %s" % str(e))
+ return self.failHandshake("Internal Server Error", HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR[0])
+
+
+ ## build response to complete WebSocket handshake
+ ##
+ response = "HTTP/1.1 %d Switching Protocols\x0d\x0a" % HTTP_STATUS_CODE_SWITCHING_PROTOCOLS[0]
+
+ if self.factory.server is not None and self.factory.server != "":
+ response += "Server: %s\x0d\x0a" % self.factory.server.encode("utf-8")
+
+ response += "Upgrade: WebSocket\x0d\x0a"
+ response += "Connection: Upgrade\x0d\x0a"
+
+ if self.websocket_protocol_in_use is not None:
+ response += "Sec-WebSocket-Protocol: %s\x0d\x0a" % str(self.websocket_protocol_in_use)
+
+ if self.websocket_version == 0:
+
+ if self.websocket_origin:
+ ## browser client provide the header, and expect it to be echo'ed
+ response += "Sec-WebSocket-Origin: %s\x0d\x0a" % str(self.websocket_origin)
+
+ if self.debugCodePaths:
+ log.msg('factory isSecure = %s port = %s' % (self.factory.isSecure, self.factory.port))
+
+ if (self.factory.isSecure and self.factory.port != 443) or ((not self.factory.isSecure) and self.factory.port != 80):
+ if self.debugCodePaths:
+ log.msg('factory running on non-default port')
+ response_port = ':' + str(self.factory.port)
+ else:
+ if self.debugCodePaths:
+ log.msg('factory running on default port')
+ response_port = ''
+
+ ## FIXME: check this! But see below ..
+ if False:
+ response_host = str(self.factory.host)
+ response_path = str(self.factory.path)
+ else:
+ response_host = str(self.http_request_host)
+ response_path = str(self.http_request_uri)
+
+ location = "%s://%s%s%s" % ('wss' if self.factory.isSecure else 'ws', response_host, response_port, response_path)
+
+ # Safari is very picky about this one
+ response += "Sec-WebSocket-Location: %s\x0d\x0a" % location
+
+ ## end of HTTP response headers
+ response += "\x0d\x0a"
+
+ ## compute accept body
+ ##
+ accept_val = struct.pack(">II", key1, key2) + key3
+ accept = hashlib.md5(accept_val).digest()
+ response_body = str(accept)
+ else:
+ ## compute Sec-WebSocket-Accept
+ ##
+ sha1 = hashlib.sha1()
+ sha1.update(key + WebSocketProtocol.WS_MAGIC)
+ sec_websocket_accept = base64.b64encode(sha1.digest())
+
+ response += "Sec-WebSocket-Accept: %s\x0d\x0a" % sec_websocket_accept
+
+ if len(self.websocket_extensions_in_use) > 0:
+ response += "Sec-WebSocket-Extensions: %s\x0d\x0a" % ','.join(self.websocket_extensions_in_use)
+
+ ## end of HTTP response headers
+ response += "\x0d\x0a"
+ response_body = ''
+
+ if self.debug:
+ log.msg("sending HTTP response:\n\n%s%s\n\n" % (response, binascii.b2a_hex(response_body)))
+
+ ## save and send out opening HS data
+ ##
+ self.http_response_data = response + response_body
+ self.sendData(self.http_response_data)
+
+ ## opening handshake completed, move WebSockets connection into OPEN state
+ ##
+ self.state = WebSocketProtocol.STATE_OPEN
+
+ ## cancel any opening HS timer if present
+ ##
+ if self.openHandshakeTimeoutCall is not None:
+ if self.debugCodePaths:
+ log.msg("openHandshakeTimeoutCall.cancel")
+ self.openHandshakeTimeoutCall.cancel()
+ self.openHandshakeTimeoutCall = None
+
+ ## init state
+ ##
+ self.inside_message = False
+ if self.websocket_version != 0:
+ self.current_frame = None
+
+ ## fire handler on derived class
+ ##
+ self.onOpen()
+
+ ## process rest, if any
+ ##
+ if len(self.data) > 0:
+ self.consumeData()
+
+
+ def failHandshake(self, reason, code = HTTP_STATUS_CODE_BAD_REQUEST[0], responseHeaders = []):
+ """
+ During opening handshake the client request was invalid, we send a HTTP
+ error response and then drop the connection.
+ """
+ if self.debug:
+ log.msg("failing WebSockets opening handshake ('%s')" % reason)
+ self.sendHttpErrorResponse(code, reason, responseHeaders)
+ self.dropConnection(abort = False)
+
+
+ def sendHttpErrorResponse(self, code, reason, responseHeaders = []):
+ """
+ Send out HTTP error response.
+ """
+ response = "HTTP/1.1 %d %s\x0d\x0a" % (code, reason.encode("utf-8"))
+ for h in responseHeaders:
+ response += "%s: %s\x0d\x0a" % (h[0], h[1].encode("utf-8"))
+ response += "\x0d\x0a"
+ self.sendData(response)
+
+
+ def sendHtml(self, html):
+ raw = html.encode("utf-8")
+ response = "HTTP/1.1 %d %s\x0d\x0a" % (HTTP_STATUS_CODE_OK[0], HTTP_STATUS_CODE_OK[1])
+ if self.factory.server is not None and self.factory.server != "":
+ response += "Server: %s\x0d\x0a" % self.factory.server.encode("utf-8")
+ response += "Content-Type: text/html; charset=UTF-8\x0d\x0a"
+ response += "Content-Length: %d\x0d\x0a" % len(raw)
+ response += "\x0d\x0a"
+ response += raw
+ self.sendData(response)
+
+
+ def sendServerStatus(self):
+ """
+ Used to send out server status/version upon receiving a HTTP/GET without
+ upgrade to WebSocket header (and option serverStatus is True).
+ """
+ html = """
+<!DOCTYPE html>
+<html>
+ <body>
+ <h1>Autobahn WebSockets %s</h1>
+ <p>
+ I am not Web server, but a WebSocket endpoint.
+ You can talk to me using the WebSocket <a href="http://tools.ietf.org/html/rfc6455">protocol</a>.
+ </p>
+ <p>
+ For more information, please visit <a href="http://autobahn.ws">my homepage</a>.
+ </p>
+ </body>
+</html>
+""" % str(autobahn.version)
+ self.sendHtml(html)
+
+
+class WebSocketServerFactory(protocol.ServerFactory, WebSocketFactory):
+ """
+ A Twisted factory for WebSockets server protocols.
+ """
+
+ protocol = WebSocketServerProtocol
+ """
+ The protocol to be spoken. Must be derived from :class:`autobahn.websocket.WebSocketServerProtocol`.
+ """
+
+
+ def __init__(self,
+
+ ## WebSockect session parameters
+ url = None,
+ protocols = [],
+ server = "AutobahnPython/%s" % autobahn.version,
+
+ ## debugging
+ debug = False,
+ debugCodePaths = False):
+ """
+ Create instance of WebSocket server factory.
+
+ Note that you MUST set URL either here or using setSessionParameters() _before_ the factory is started.
+
+ :param url: WebSocket listening URL - ("ws:" | "wss:") "//" host [ ":" port ] path [ "?" query ].
+ :type url: str
+ :param protocols: List of subprotocols the server supports. The subprotocol used is the first from the list of subprotocols announced by the client that is contained in this list.
+ :type protocols: list of strings
+ :param server: Server as announced in HTTP response header during opening handshake or None (default: "AutobahnWebSockets/x.x.x").
+ :type server: str
+ :param debug: Debug mode (default: False).
+ :type debug: bool
+ :param debugCodePaths: Debug code paths mode (default: False).
+ :type debugCodePaths: bool
+ """
+ self.debug = debug
+ self.debugCodePaths = debugCodePaths
+
+ self.logOctets = debug
+ self.logFrames = debug
+
+ self.isServer = True
+
+ ## seed RNG which is used for WS frame masks generation
+ random.seed()
+
+ ## default WS session parameters
+ ##
+ self.setSessionParameters(url, protocols, server)
+
+ ## default WebSocket protocol options
+ ##
+ self.resetProtocolOptions()
+
+ ## number of currently connected clients
+ ##
+ self.countConnections = 0
+
+
+ def setSessionParameters(self, url = None, protocols = [], server = None):
+ """
+ Set WebSocket session parameters.
+
+ :param url: WebSocket listening URL - ("ws:" | "wss:") "//" host [ ":" port ].
+ :type url: str
+ :param protocols: List of subprotocols the server supports. The subprotocol used is the first from the list of subprotocols announced by the client that is contained in this list.
+ :type protocols: list of strings
+ :param server: Server as announced in HTTP response header during opening handshake.
+ :type server: str
+ """
+ if url is not None:
+ ## parse WebSocket URI into components
+ (isSecure, host, port, resource, path, params) = parseWsUrl(url)
+ if path != "/":
+ raise Exception("path specified for server WebSocket URL")
+ if len(params) > 0:
+ raise Exception("query parameters specified for server WebSocket URL")
+ self.url = url
+ self.isSecure = isSecure
+ self.host = host
+ self.port = port
+ else:
+ self.url = None
+ self.isSecure = None
+ self.host = None
+ self.port = None
+
+ self.protocols = protocols
+ self.server = server
+
+
+ def resetProtocolOptions(self):
+ """
+ Reset all WebSocket protocol options to defaults.
+ """
+ self.versions = WebSocketProtocol.SUPPORTED_PROTOCOL_VERSIONS
+ self.allowHixie76 = WebSocketProtocol.DEFAULT_ALLOW_HIXIE76
+ self.webStatus = True
+ self.utf8validateIncoming = True
+ self.requireMaskedClientFrames = True
+ self.maskServerFrames = False
+ self.applyMask = True
+ self.maxFramePayloadSize = 0
+ self.maxMessagePayloadSize = 0
+ self.autoFragmentSize = 0
+ self.failByDrop = True
+ self.echoCloseCodeReason = False
+ self.openHandshakeTimeout = 5
+ self.closeHandshakeTimeout = 1
+ self.tcpNoDelay = True
+
+
+ def setProtocolOptions(self,
+ versions = None,
+ allowHixie76 = None,
+ webStatus = None,
+ utf8validateIncoming = None,
+ maskServerFrames = None,
+ requireMaskedClientFrames = None,
+ applyMask = None,
+ maxFramePayloadSize = None,
+ maxMessagePayloadSize = None,
+ autoFragmentSize = None,
+ failByDrop = None,
+ echoCloseCodeReason = None,
+ openHandshakeTimeout = None,
+ closeHandshakeTimeout = None,
+ tcpNoDelay = None):
+ """
+ Set WebSocket protocol options used as defaults for new protocol instances.
+
+ :param versions: The WebSockets protocol versions accepted by the server (default: WebSocketProtocol.SUPPORTED_PROTOCOL_VERSIONS).
+ :type versions: list of ints
+ :param allowHixie76: Allow to speak Hixie76 protocol version.
+ :type allowHixie76: bool
+ :param webStatus: Return server status/version on HTTP/GET without WebSocket upgrade header (default: True).
+ :type webStatus: bool
+ :param utf8validateIncoming: Validate incoming UTF-8 in text message payloads (default: True).
+ :type utf8validateIncoming: bool
+ :param maskServerFrames: Mask server-to-client frames (default: False).
+ :type maskServerFrames: bool
+ :param requireMaskedClientFrames: Require client-to-server frames to be masked (default: True).
+ :type requireMaskedClientFrames: bool
+ :param applyMask: Actually apply mask to payload when mask it present. Applies for outgoing and incoming frames (default: True).
+ :type applyMask: bool
+ :param maxFramePayloadSize: Maximum frame payload size that will be accepted when receiving or 0 for unlimited (default: 0).
+ :type maxFramePayloadSize: int
+ :param maxMessagePayloadSize: Maximum message payload size (after reassembly of fragmented messages) that will be accepted when receiving or 0 for unlimited (default: 0).
+ :type maxMessagePayloadSize: int
+ :param autoFragmentSize: Automatic fragmentation of outgoing data messages (when using the message-based API) into frames with payload length <= this size or 0 for no auto-fragmentation (default: 0).
+ :type autoFragmentSize: int
+ :param failByDrop: Fail connections by dropping the TCP connection without performaing closing handshake (default: True).
+ :type failbyDrop: bool
+ :param echoCloseCodeReason: Iff true, when receiving a close, echo back close code/reason. Otherwise reply with code == NORMAL, reason = "" (default: False).
+ :type echoCloseCodeReason: bool
+ :param openHandshakeTimeout: Opening WebSocket handshake timeout, timeout in seconds or 0 to deactivate (default: 0).
+ :type openHandshakeTimeout: float
+ :param closeHandshakeTimeout: When we expect to receive a closing handshake reply, timeout in seconds (default: 1).
+ :type closeHandshakeTimeout: float
+ :param tcpNoDelay: TCP NODELAY ("Nagle") socket option (default: True).
+ :type tcpNoDelay: bool
+ """
+ if allowHixie76 is not None and allowHixie76 != self.allowHixie76:
+ self.allowHixie76 = allowHixie76
+
+ if versions is not None:
+ for v in versions:
+ if v not in WebSocketProtocol.SUPPORTED_PROTOCOL_VERSIONS:
+ raise Exception("invalid WebSockets protocol version %s (allowed values: %s)" % (v, str(WebSocketProtocol.SUPPORTED_PROTOCOL_VERSIONS)))
+ if v == 0 and not self.allowHixie76:
+ raise Exception("use of Hixie-76 requires allowHixie76 == True")
+ if set(versions) != set(self.versions):
+ self.versions = versions
+
+ if webStatus is not None and webStatus != self.webStatus:
+ self.webStatus = webStatus
+
+ if utf8validateIncoming is not None and utf8validateIncoming != self.utf8validateIncoming:
+ self.utf8validateIncoming = utf8validateIncoming
+
+ if requireMaskedClientFrames is not None and requireMaskedClientFrames != self.requireMaskedClientFrames:
+ self.requireMaskedClientFrames = requireMaskedClientFrames
+
+ if maskServerFrames is not None and maskServerFrames != self.maskServerFrames:
+ self.maskServerFrames = maskServerFrames
+
+ if applyMask is not None and applyMask != self.applyMask:
+ self.applyMask = applyMask
+
+ if maxFramePayloadSize is not None and maxFramePayloadSize != self.maxFramePayloadSize:
+ self.maxFramePayloadSize = maxFramePayloadSize
+
+ if maxMessagePayloadSize is not None and maxMessagePayloadSize != self.maxMessagePayloadSize:
+ self.maxMessagePayloadSize = maxMessagePayloadSize
+
+ if autoFragmentSize is not None and autoFragmentSize != self.autoFragmentSize:
+ self.autoFragmentSize = autoFragmentSize
+
+ if failByDrop is not None and failByDrop != self.failByDrop:
+ self.failByDrop = failByDrop
+
+ if echoCloseCodeReason is not None and echoCloseCodeReason != self.echoCloseCodeReason:
+ self.echoCloseCodeReason = echoCloseCodeReason
+
+ if openHandshakeTimeout is not None and openHandshakeTimeout != self.openHandshakeTimeout:
+ self.openHandshakeTimeout = openHandshakeTimeout
+
+ if closeHandshakeTimeout is not None and closeHandshakeTimeout != self.closeHandshakeTimeout:
+ self.closeHandshakeTimeout = closeHandshakeTimeout
+
+ if tcpNoDelay is not None and tcpNoDelay != self.tcpNoDelay:
+ self.tcpNoDelay = tcpNoDelay
+
+
+ def getConnectionCount(self):
+ """
+ Get number of currently connected clients.
+
+ :returns: int -- Number of currently connected clients.
+ """
+ return self.countConnections
+
+
+ def startFactory(self):
+ """
+ Called by Twisted before starting to listen on port for incoming connections.
+ Default implementation does nothing. Override in derived class when appropriate.
+ """
+ pass
+
+
+ def stopFactory(self):
+ """
+ Called by Twisted before stopping to listen on port for incoming connections.
+ Default implementation does nothing. Override in derived class when appropriate.
+ """
+ pass
+
+
+class WebSocketClientProtocol(WebSocketProtocol):
+ """
+ Client protocol for WebSockets.
+ """
+
+ def onConnect(self, connectionResponse):
+ """
+ Callback fired directly after WebSocket opening handshake when new WebSocket server
+ connection was established.
+
+ :param connectionResponse: WebSocket connection response information.
+ :type connectionResponse: instance of :class:`autobahn.websocket.ConnectionResponse`
+ """
+ pass
+
+
+ def connectionMade(self):
+ """
+ Called by Twisted when new TCP connection to server was established. Default
+ implementation will start the initial WebSocket opening handshake.
+ When overriding in derived class, make sure to call this base class
+ implementation _before_ your code.
+ """
+ self.isServer = False
+ WebSocketProtocol.connectionMade(self)
+ if self.debug:
+ log.msg("connection to %s established" % self.peerstr)
+ self.startHandshake()
+
+
+ def connectionLost(self, reason):
+ """
+ Called by Twisted when established TCP connection to server was lost. Default
+ implementation will tear down all state properly.
+ When overriding in derived class, make sure to call this base class
+ implementation _after_ your code.
+ """
+ WebSocketProtocol.connectionLost(self, reason)
+ if self.debug:
+ log.msg("connection to %s lost" % self.peerstr)
+
+
+ def createHixieKey(self):
+ """
+ Supposed to implement the crack smoker algorithm below. Well, crack
+ probably wasn't the stuff they smoked - dog poo?
+
+ http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-76#page-21
+ Items 16 - 22
+ """
+ spaces1 = random.randint(1, 12)
+ max1 = int(4294967295L / spaces1)
+ number1 = random.randint(0, max1)
+ product1 = number1 * spaces1
+ key1 = str(product1)
+ rchars = filter(lambda x: (x >= 0x21 and x <= 0x2f) or (x >= 0x3a and x <= 0x7e), range(0,127))
+ for i in xrange(random.randint(1, 12)):
+ p = random.randint(0, len(key1) - 1)
+ key1 = key1[:p] + chr(random.choice(rchars)) + key1[p:]
+ for i in xrange(spaces1):
+ p = random.randint(1, len(key1) - 2)
+ key1 = key1[:p] + ' ' + key1[p:]
+ return (key1, number1)
+
+
+ def startHandshake(self):
+ """
+ Start WebSockets opening handshake.
+ """
+
+ ## construct WS opening handshake HTTP header
+ ##
+ request = "GET %s HTTP/1.1\x0d\x0a" % self.factory.resource.encode("utf-8")
+
+ if self.factory.useragent is not None and self.factory.useragent != "":
+ request += "User-Agent: %s\x0d\x0a" % self.factory.useragent.encode("utf-8")
+
+ request += "Host: %s:%d\x0d\x0a" % (self.factory.host.encode("utf-8"), self.factory.port)
+ request += "Upgrade: WebSocket\x0d\x0a"
+ request += "Connection: Upgrade\x0d\x0a"
+
+ ## handshake random key
+ ##
+ if self.version == 0:
+ (self.websocket_key1, number1) = self.createHixieKey()
+ (self.websocket_key2, number2) = self.createHixieKey()
+ self.websocket_key3 = os.urandom(8)
+ accept_val = struct.pack(">II", number1, number2) + self.websocket_key3
+ self.websocket_expected_challenge_response = hashlib.md5(accept_val).digest()
+
+ request += "Sec-WebSocket-Key1: %s\x0d\x0a" % self.websocket_key1
+ request += "Sec-WebSocket-Key2: %s\x0d\x0a" % self.websocket_key2
+ else:
+ self.websocket_key = base64.b64encode(os.urandom(16))
+ request += "Sec-WebSocket-Key: %s\x0d\x0a" % self.websocket_key
+
+ ## optional origin announced
+ ##
+ if self.factory.origin:
+ if self.version > 10 or self.version == 0:
+ request += "Origin: %d\x0d\x0a" % self.factory.origin.encode("utf-8")
+ else:
+ request += "Sec-WebSocket-Origin: %d\x0d\x0a" % self.factory.origin.encode("utf-8")
+
+ ## optional list of WS subprotocols announced
+ ##
+ if len(self.factory.protocols) > 0:
+ request += "Sec-WebSocket-Protocol: %s\x0d\x0a" % ','.join(self.factory.protocols)
+
+ ## set WS protocol version depending on WS spec version
+ ##
+ if self.version != 0:
+ request += "Sec-WebSocket-Version: %d\x0d\x0a" % WebSocketProtocol.SPEC_TO_PROTOCOL_VERSION[self.version]
+
+ request += "\x0d\x0a"
+
+ if self.version == 0:
+ request += self.websocket_key3
+
+ self.http_request_data = request
+
+ if self.debug:
+ log.msg(self.http_request_data)
+
+ self.sendData(self.http_request_data)
+
+
+ def processHandshake(self):
+ """
+ Process WebSockets opening handshake response from server.
+ """
+ ## only proceed when we have fully received the HTTP request line and all headers
+ ##
+ end_of_header = self.data.find("\x0d\x0a\x0d\x0a")
+ if end_of_header >= 0:
+
+ self.http_response_data = self.data[:end_of_header + 4]
+ if self.debug:
+ log.msg("received HTTP response:\n\n%s\n\n" % self.http_response_data)
+
+ ## extract HTTP status line and headers
+ ##
+ (self.http_status_line, self.http_headers, http_headers_cnt) = parseHttpHeader(self.http_response_data)
+
+ ## validate WebSocket opening handshake server response
+ ##
+ if self.debug:
+ log.msg("received HTTP status line in opening handshake : %s" % str(self.http_status_line))
+ log.msg("received HTTP headers in opening handshake : %s" % str(self.http_headers))
+
+ ## Response Line
+ ##
+ sl = self.http_status_line.split()
+ if len(sl) < 2:
+ return self.failHandshake("Bad HTTP response status line '%s'" % self.http_status_line)
+
+ ## HTTP version
+ ##
+ http_version = sl[0].strip()
+ if http_version != "HTTP/1.1":
+ return self.failHandshake("Unsupported HTTP version ('%s')" % http_version)
+
+ ## HTTP status code
+ ##
+ try:
+ status_code = int(sl[1].strip())
+ except:
+ return self.failHandshake("Bad HTTP status code ('%s')" % sl[1].strip())
+ if status_code != HTTP_STATUS_CODE_SWITCHING_PROTOCOLS[0]:
+
+ ## FIXME: handle redirects
+ ## FIXME: handle authentication required
+
+ if len(sl) > 2:
+ reason = " - %s" % sl[2].strip()
+ else:
+ reason = ""
+ return self.failHandshake("WebSockets connection upgrade failed (%d%s)" % (status_code, reason))
+
+ ## Upgrade
+ ##
+ if not self.http_headers.has_key("upgrade"):
+ return self.failHandshake("HTTP Upgrade header missing")
+ if self.http_headers["upgrade"].strip().lower() != "websocket":
+ return self.failHandshake("HTTP Upgrade header different from 'websocket' (case-insensitive) : %s" % self.http_headers["upgrade"])
+
+ ## Connection
+ ##
+ if not self.http_headers.has_key("connection"):
+ return self.failHandshake("HTTP Connection header missing")
+ connectionUpgrade = False
+ for c in self.http_headers["connection"].split(","):
+ if c.strip().lower() == "upgrade":
+ connectionUpgrade = True
+ break
+ if not connectionUpgrade:
+ return self.failHandshake("HTTP Connection header does not include 'upgrade' value (case-insensitive) : %s" % self.http_headers["connection"])
+
+ ## compute Sec-WebSocket-Accept
+ ##
+ if self.version != 0:
+ if not self.http_headers.has_key("sec-websocket-accept"):
+ return self.failHandshake("HTTP Sec-WebSocket-Accept header missing in opening handshake reply")
+ else:
+ if http_headers_cnt["sec-websocket-accept"] > 1:
+ return self.failHandshake("HTTP Sec-WebSocket-Accept header appears more than once in opening handshake reply")
+ sec_websocket_accept_got = self.http_headers["sec-websocket-accept"].strip()
+
+ sha1 = hashlib.sha1()
+ sha1.update(self.websocket_key + WebSocketProtocol.WS_MAGIC)
+ sec_websocket_accept = base64.b64encode(sha1.digest())
+
+ if sec_websocket_accept_got != sec_websocket_accept:
+ return self.failHandshake("HTTP Sec-WebSocket-Accept bogus value : expected %s / got %s" % (sec_websocket_accept, sec_websocket_accept_got))
+
+ ## handle "extensions in use" - if any
+ ##
+ self.websocket_extensions_in_use = []
+ if self.version != 0:
+ if self.http_headers.has_key("sec-websocket-extensions"):
+ if http_headers_cnt["sec-websocket-extensions"] > 1:
+ return self.failHandshake("HTTP Sec-WebSocket-Extensions header appears more than once in opening handshake reply")
+ exts = self.http_headers["sec-websocket-extensions"].strip()
+ ##
+ ## we don't support any extension, but if we did, we needed
+ ## to set self.websocket_extensions_in_use here, and don't fail the handshake
+ ##
+ return self.failHandshake("server wants to use extensions (%s), but no extensions implemented" % exts)
+
+ ## handle "subprotocol in use" - if any
+ ##
+ self.websocket_protocol_in_use = None
+ if self.http_headers.has_key("sec-websocket-protocol"):
+ if http_headers_cnt["sec-websocket-protocol"] > 1:
+ return self.failHandshake("HTTP Sec-WebSocket-Protocol header appears more than once in opening handshake reply")
+ sp = str(self.http_headers["sec-websocket-protocol"].strip())
+ if sp != "":
+ if sp not in self.factory.protocols:
+ return self.failHandshake("subprotocol selected by server (%s) not in subprotocol list requested by client (%s)" % (sp, str(self.factory.protocols)))
+ else:
+ ## ok, subprotocol in use
+ ##
+ self.websocket_protocol_in_use = sp
+
+
+ ## For Hixie-76, we need 16 octets of HTTP request body to complete HS!
+ ##
+ if self.version == 0:
+ if len(self.data) < end_of_header + 4 + 16:
+ return
+ else:
+ challenge_response = self.data[end_of_header + 4:end_of_header + 4 + 16]
+ if challenge_response != self.websocket_expected_challenge_response:
+ return self.failHandshake("invalid challenge response received from server (Hixie-76)")
+
+ ## Ok, got complete HS input, remember rest (if any)
+ ##
+ if self.version == 0:
+ self.data = self.data[end_of_header + 4 + 16:]
+ else:
+ self.data = self.data[end_of_header + 4:]
+
+ ## opening handshake completed, move WebSockets connection into OPEN state
+ ##
+ self.state = WebSocketProtocol.STATE_OPEN
+ self.inside_message = False
+ if self.version != 0:
+ self.current_frame = None
+ self.websocket_version = self.version
+
+ ## we handle this symmetrical to server-side .. that is, give the
+ ## client a chance to bail out .. i.e. on no subprotocol selected
+ ## by server
+ try:
+ connectionResponse = ConnectionResponse(self.peer,
+ self.peerstr,
+ self.http_headers,
+ None, # FIXME
+ self.websocket_protocol_in_use,
+ self.websocket_extensions_in_use)
+
+ self.onConnect(connectionResponse)
+
+ except Exception, e:
+ ## immediately close the WS connection
+ ##
+ self.failConnection(1000, str(e))
+ else:
+ ## fire handler on derived class
+ ##
+ self.onOpen()
+
+ ## process rest, if any
+ ##
+ if len(self.data) > 0:
+ self.consumeData()
+
+
+ def failHandshake(self, reason):
+ """
+ During opening handshake the server response is invalid and we drop the
+ connection.
+ """
+ if self.debug:
+ log.msg("failing WebSockets opening handshake ('%s')" % reason)
+ self.dropConnection(abort = True)
+
+
+class WebSocketClientFactory(protocol.ClientFactory, WebSocketFactory):
+ """
+ A Twisted factory for WebSockets client protocols.
+ """
+
+ protocol = WebSocketClientProtocol
+ """
+ The protocol to be spoken. Must be derived from :class:`autobahn.websocket.WebSocketClientProtocol`.
+ """
+
+
+ def __init__(self,
+
+ ## WebSockect session parameters
+ url = None,
+ origin = None,
+ protocols = [],
+ useragent = "AutobahnPython/%s" % autobahn.version,
+
+ ## debugging
+ debug = False,
+ debugCodePaths = False):
+ """
+ Create instance of WebSocket client factory.
+
+ Note that you MUST set URL either here or using setSessionParameters() _before_ the factory is started.
+
+ :param url: WebSocket URL to connect to - ("ws:" | "wss:") "//" host [ ":" port ] path [ "?" query ].
+ :type url: str
+ :param origin: The origin to be sent in WebSockets opening handshake or None (default: None).
+ :type origin: str
+ :param protocols: List of subprotocols the client should announce in WebSockets opening handshake (default: []).
+ :type protocols: list of strings
+ :param useragent: User agent as announced in HTTP request header or None (default: "AutobahnWebSockets/x.x.x").
+ :type useragent: str
+ :param debug: Debug mode (default: False).
+ :type debug: bool
+ :param debugCodePaths: Debug code paths mode (default: False).
+ :type debugCodePaths: bool
+ """
+ self.debug = debug
+ self.debugCodePaths = debugCodePaths
+
+ self.logOctets = debug
+ self.logFrames = debug
+
+ self.isServer = False
+
+ ## seed RNG which is used for WS opening handshake key and WS frame masks generation
+ random.seed()
+
+ ## default WS session parameters
+ ##
+ self.setSessionParameters(url, origin, protocols, useragent)
+
+ ## default WebSocket protocol options
+ ##
+ self.resetProtocolOptions()
+
+
+ def setSessionParameters(self, url = None, origin = None, protocols = [], useragent = None):
+ """
+ Set WebSocket session parameters.
+
+ :param url: WebSocket URL to connect to - ("ws:" | "wss:") "//" host [ ":" port ] path [ "?" query ].
+ :type url: str
+ :param origin: The origin to be sent in opening handshake.
+ :type origin: str
+ :param protocols: List of WebSocket subprotocols the client should announce in opening handshake.
+ :type protocols: list of strings
+ :param useragent: User agent as announced in HTTP request header during opening handshake.
+ :type useragent: str
+ """
+ if url is not None:
+ ## parse WebSocket URI into components
+ (isSecure, host, port, resource, path, params) = parseWsUrl(url)
+ self.url = url
+ self.isSecure = isSecure
+ self.host = host
+ self.port = port
+ self.resource = resource
+ self.path = path
+ self.params = params
+ else:
+ self.url = None
+ self.isSecure = None
+ self.host = None
+ self.port = None
+ self.resource = None
+ self.path = None
+ self.params = None
+
+ self.origin = origin
+ self.protocols = protocols
+ self.useragent = useragent
+
+
+ def resetProtocolOptions(self):
+ """
+ Reset all WebSocket protocol options to defaults.
+ """
+ self.version = WebSocketProtocol.DEFAULT_SPEC_VERSION
+ self.allowHixie76 = WebSocketProtocol.DEFAULT_ALLOW_HIXIE76
+ self.utf8validateIncoming = True
+ self.acceptMaskedServerFrames = False
+ self.maskClientFrames = True
+ self.applyMask = True
+ self.maxFramePayloadSize = 0
+ self.maxMessagePayloadSize = 0
+ self.autoFragmentSize = 0
+ self.failByDrop = True
+ self.echoCloseCodeReason = False
+ self.serverConnectionDropTimeout = 1
+ self.openHandshakeTimeout = 5
+ self.closeHandshakeTimeout = 1
+ self.tcpNoDelay = True
+
+
+ def setProtocolOptions(self,
+ version = None,
+ allowHixie76 = None,
+ utf8validateIncoming = None,
+ acceptMaskedServerFrames = None,
+ maskClientFrames = None,
+ applyMask = None,
+ maxFramePayloadSize = None,
+ maxMessagePayloadSize = None,
+ autoFragmentSize = None,
+ failByDrop = None,
+ echoCloseCodeReason = None,
+ serverConnectionDropTimeout = None,
+ openHandshakeTimeout = None,
+ closeHandshakeTimeout = None,
+ tcpNoDelay = None):
+ """
+ Set WebSocket protocol options used as defaults for _new_ protocol instances.
+
+ :param version: The WebSockets protocol spec (draft) version to be used (default: WebSocketProtocol.DEFAULT_SPEC_VERSION).
+ :type version: int
+ :param allowHixie76: Allow to speak Hixie76 protocol version.
+ :type allowHixie76: bool
+ :param utf8validateIncoming: Validate incoming UTF-8 in text message payloads (default: True).
+ :type utf8validateIncoming: bool
+ :param acceptMaskedServerFrames: Accept masked server-to-client frames (default: False).
+ :type acceptMaskedServerFrames: bool
+ :param maskClientFrames: Mask client-to-server frames (default: True).
+ :type maskClientFrames: bool
+ :param applyMask: Actually apply mask to payload when mask it present. Applies for outgoing and incoming frames (default: True).
+ :type applyMask: bool
+ :param maxFramePayloadSize: Maximum frame payload size that will be accepted when receiving or 0 for unlimited (default: 0).
+ :type maxFramePayloadSize: int
+ :param maxMessagePayloadSize: Maximum message payload size (after reassembly of fragmented messages) that will be accepted when receiving or 0 for unlimited (default: 0).
+ :type maxMessagePayloadSize: int
+ :param autoFragmentSize: Automatic fragmentation of outgoing data messages (when using the message-based API) into frames with payload length <= this size or 0 for no auto-fragmentation (default: 0).
+ :type autoFragmentSize: int
+ :param failByDrop: Fail connections by dropping the TCP connection without performing closing handshake (default: True).
+ :type failbyDrop: bool
+ :param echoCloseCodeReason: Iff true, when receiving a close, echo back close code/reason. Otherwise reply with code == NORMAL, reason = "" (default: False).
+ :type echoCloseCodeReason: bool
+ :param serverConnectionDropTimeout: When the client expects the server to drop the TCP, timeout in seconds (default: 1).
+ :type serverConnectionDropTimeout: float
+ :param openHandshakeTimeout: Opening WebSocket handshake timeout, timeout in seconds or 0 to deactivate (default: 0).
+ :type openHandshakeTimeout: float
+ :param closeHandshakeTimeout: When we expect to receive a closing handshake reply, timeout in seconds (default: 1).
+ :type closeHandshakeTimeout: float
+ :param tcpNoDelay: TCP NODELAY ("Nagle") socket option (default: True).
+ :type tcpNoDelay: bool
+ """
+ if allowHixie76 is not None and allowHixie76 != self.allowHixie76:
+ self.allowHixie76 = allowHixie76
+
+ if version is not None:
+ if version not in WebSocketProtocol.SUPPORTED_SPEC_VERSIONS:
+ raise Exception("invalid WebSockets draft version %s (allowed values: %s)" % (version, str(WebSocketProtocol.SUPPORTED_SPEC_VERSIONS)))
+ if version == 0 and not self.allowHixie76:
+ raise Exception("use of Hixie-76 requires allowHixie76 == True")
+ if version != self.version:
+ self.version = version
+
+ if utf8validateIncoming is not None and utf8validateIncoming != self.utf8validateIncoming:
+ self.utf8validateIncoming = utf8validateIncoming
+
+ if acceptMaskedServerFrames is not None and acceptMaskedServerFrames != self.acceptMaskedServerFrames:
+ self.acceptMaskedServerFrames = acceptMaskedServerFrames
+
+ if maskClientFrames is not None and maskClientFrames != self.maskClientFrames:
+ self.maskClientFrames = maskClientFrames
+
+ if applyMask is not None and applyMask != self.applyMask:
+ self.applyMask = applyMask
+
+ if maxFramePayloadSize is not None and maxFramePayloadSize != self.maxFramePayloadSize:
+ self.maxFramePayloadSize = maxFramePayloadSize
+
+ if maxMessagePayloadSize is not None and maxMessagePayloadSize != self.maxMessagePayloadSize:
+ self.maxMessagePayloadSize = maxMessagePayloadSize
+
+ if autoFragmentSize is not None and autoFragmentSize != self.autoFragmentSize:
+ self.autoFragmentSize = autoFragmentSize
+
+ if failByDrop is not None and failByDrop != self.failByDrop:
+ self.failByDrop = failByDrop
+
+ if echoCloseCodeReason is not None and echoCloseCodeReason != self.echoCloseCodeReason:
+ self.echoCloseCodeReason = echoCloseCodeReason
+
+ if serverConnectionDropTimeout is not None and serverConnectionDropTimeout != self.serverConnectionDropTimeout:
+ self.serverConnectionDropTimeout = serverConnectionDropTimeout
+
+ if openHandshakeTimeout is not None and openHandshakeTimeout != self.openHandshakeTimeout:
+ self.openHandshakeTimeout = openHandshakeTimeout
+
+ if closeHandshakeTimeout is not None and closeHandshakeTimeout != self.closeHandshakeTimeout:
+ self.closeHandshakeTimeout = closeHandshakeTimeout
+
+ if tcpNoDelay is not None and tcpNoDelay != self.tcpNoDelay:
+ self.tcpNoDelay = tcpNoDelay
+
+
+ def clientConnectionFailed(self, connector, reason):
+ """
+ Called by Twisted when the connection to server has failed. Default implementation
+ does nothing. Override in derived class when appropriate.
+ """
+ pass
+
+
+ def clientConnectionLost(self, connector, reason):
+ """
+ Called by Twisted when the connection to server was lost. Default implementation
+ does nothing. Override in derived class when appropriate.
+ """
+ pass
diff --git a/autobahn/xormasker.py b/autobahn/xormasker.py new file mode 100644 index 0000000..b480727 --- /dev/null +++ b/autobahn/xormasker.py @@ -0,0 +1,85 @@ +###############################################################################
+##
+## Copyright 2012 Tavendo GmbH
+##
+## Licensed under the Apache License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+##
+###############################################################################
+
+from array import array
+
+
+class XorMaskerNull:
+
+ def __init__(self, mask = None):
+ self.ptr = 0
+
+ def pointer(self):
+ return self.ptr
+
+ def reset(self):
+ self.ptr = 0
+
+ def process(self, data):
+ self.ptr += len(data)
+ return data
+
+
+class XorMaskerSimple:
+
+ def __init__(self, mask):
+ assert len(mask) == 4
+ self.ptr = 0
+ self.msk = array('B', mask)
+
+ def pointer(self):
+ return self.ptr
+
+ def reset(self):
+ self.ptr = 0
+
+ def process(self, data):
+ dlen = len(data)
+ payload = array('B', data)
+ for k in xrange(dlen):
+ payload[k] ^= self.msk[self.ptr & 3]
+ self.ptr += 1
+ return payload.tostring()
+
+
+class XorMaskerShifted1:
+
+ def __init__(self, mask):
+ assert len(mask) == 4
+ self.ptr = 0
+ self.mskarray = [array('B'), array('B'), array('B'), array('B')]
+ for j in xrange(4):
+ self.mskarray[0].append(ord(mask[ j & 3]))
+ self.mskarray[1].append(ord(mask[(j + 1) & 3]))
+ self.mskarray[2].append(ord(mask[(j + 2) & 3]))
+ self.mskarray[3].append(ord(mask[(j + 3) & 3]))
+
+ def pointer(self):
+ return self.ptr
+
+ def reset(self):
+ self.ptr = 0
+
+ def process(self, data):
+ dlen = len(data)
+ payload = array('B', data)
+ msk = self.mskarray[self.ptr & 3]
+ for k in xrange(dlen):
+ payload[k] ^= msk[k & 3]
+ self.ptr += dlen
+ return payload.tostring()
|