diff --git a/xcap/appusage/__init__.py b/xcap/appusage/__init__.py index 2bb90b7..5d5390b 100644 --- a/xcap/appusage/__init__.py +++ b/xcap/appusage/__init__.py @@ -1,377 +1,379 @@ # Copyright (C) 2007-2010 AG-Projects. # """XCAP application usage module""" import os import sys from cStringIO import StringIO from lxml import etree from application.configuration import ConfigSection, ConfigSetting from application.configuration.datatypes import StringList from application import log import xcap from xcap import errors from xcap import element from xcap.interfaces.backend import StatusResponse class Backend(object): """Configuration datatype, used to select a backend module from the configuration file.""" def __new__(typ, value): value = value.lower() try: return __import__('xcap.interfaces.backend.%s' % value, globals(), locals(), ['']) except (ImportError, AssertionError), e: log.fatal("Cannot load '%s' backend module: %s" % (value, str(e))) sys.exit(1) except Exception, e: log.err() sys.exit(1) class ServerConfig(ConfigSection): __cfgfile__ = xcap.__cfgfile__ __section__ = 'Server' backend = ConfigSetting(type=Backend, value=None) disabled_applications = ConfigSetting(type=StringList, value=[]) document_validation = True if ServerConfig.backend is None: log.fatal("OpenXCAP needs a backend to be specified in order to run") sys.exit(1) class ApplicationUsage(object): """Base class defining an XCAP application""" id = None ## the Application Unique ID (AUID) default_ns = None ## the default XML namespace mime_type = None ## the MIME type schema_file = None ## filename of the schema for the application def __init__(self, storage): ## the XML schema that defines valid documents for this application if self.schema_file: xml_schema_doc = etree.parse(open(os.path.join(os.path.dirname(__file__), 'xml-schemas', self.schema_file), 'r')) self.xml_schema = etree.XMLSchema(xml_schema_doc) else: class EverythingIsValid(object): def __call__(self, *args, **kw): return True def validate(self, *args, **kw): return True self.xml_schema = EverythingIsValid() if storage is not None: self.storage = storage ## Validation def _check_UTF8_encoding(self, xml_doc): """Check if the document is UTF8 encoded. Raise an NotUTF8Error if it's not.""" if xml_doc.docinfo.encoding.lower() != 'utf-8': raise errors.NotUTF8Error(comment='document encoding is %s' % xml_doc.docinfo.encoding) def _check_schema_validation(self, xml_doc): """Check if the given XCAP document validates against the application's schema""" if not self.xml_schema(xml_doc): raise errors.SchemaValidationError(comment=self.xml_schema.error_log) def _check_additional_constraints(self, xml_doc): """Check additional validations constraints for this XCAP document. Should be overriden in subclasses if specified by the application usage, and raise a ConstraintFailureError if needed.""" def validate_document(self, xcap_doc): """Check if a document is valid for this application.""" try: xml_doc = etree.parse(StringIO(xcap_doc)) # XXX do not use TreeBuilder here except etree.XMLSyntaxError, ex: ex.http_error = errors.NotWellFormedError(comment=str(ex)) raise except Exception, ex: ex.http_error = errors.NotWellFormedError() raise self._check_UTF8_encoding(xml_doc) if ServerConfig.document_validation: self._check_schema_validation(xml_doc) self._check_additional_constraints(xml_doc) ## Authorization policy def is_authorized(self, xcap_user, xcap_uri): """Default authorization policy. Authorizes an XCAPUser for an XCAPUri. Return True if the user is authorized, False otherwise.""" if xcap_user and xcap_user == xcap_uri.user: return True return False ## Document management def _not_implemented(self, context): raise errors.ResourceNotFound("Application %s does not implement %s context" % (self.id, context)) def get_document(self, uri, check_etag): context = uri.doc_selector.context if context == 'global': return self.get_document_global(uri, check_etag) elif context == 'users': return self.get_document_local(uri, check_etag) else: self._not_implemented(context) def get_document_global(self, uri, check_etag): self._not_implemented('global') def get_document_local(self, uri, check_etag): return self.storage.get_document(uri, check_etag) def put_document(self, uri, document, check_etag): self.validate_document(document) return self.storage.put_document(uri, document, check_etag) def delete_document(self, uri, check_etag): return self.storage.delete_document(uri, check_etag) ## Element management def _cb_put_element(self, response, uri, element_body, check_etag): """This is called when the document that relates to the element is retrieved.""" if response.code == 404: ### XXX let the storate raise raise errors.NoParentError ### catch error in errback and attach http_error fixed_element_selector = uri.node_selector.element_selector.fix_star(element_body) try: result = element.put(response.data, fixed_element_selector, element_body) except element.SelectorError, ex: ex.http_error = errors.NoParentError(comment=str(ex)) raise if result is None: raise errors.NoParentError new_document, created = result get_result = element.get(new_document, uri.node_selector.element_selector) if get_result != element_body.strip(): raise errors.CannotInsertError('PUT request failed GET(PUT(x))==x invariant') d = self.put_document(uri, new_document, check_etag) def set_201_code(response): try: if response.code==200: response.code = 201 except AttributeError: pass return response if created: d.addCallback(set_201_code) return d def put_element(self, uri, element_body, check_etag): try: element.check_xml_fragment(element_body) except element.sax.SAXParseException, ex: ex.http_error = errors.NotXMLFragmentError(comment=str(ex)) raise except Exception, ex: ex.http_error = errors.NotXMLFragmentError() raise d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_put_element, callbackArgs=(uri, element_body, check_etag)) def _cb_get_element(self, response, uri): """This is called when the document related to the element is retrieved.""" if response.code == 404: ## XXX why not let the storage raise? raise errors.ResourceNotFound("The requested document %s was not found on this server" % uri.doc_selector) result = element.get(response.data, uri.node_selector.element_selector) if not result: msg = "The requested element %s was not found in the document %s" % (uri.node_selector, uri.doc_selector) raise errors.ResourceNotFound(msg) return StatusResponse(200, response.etag, result) def get_element(self, uri, check_etag): d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_get_element, callbackArgs=(uri, )) def _cb_delete_element(self, response, uri, check_etag): if response.code == 404: raise errors.ResourceNotFound("The requested document %s was not found on this server" % uri.doc_selector) new_document = element.delete(response.data, uri.node_selector.element_selector) if not new_document: raise errors.ResourceNotFound get_result = element.find(new_document, uri.node_selector.element_selector) if get_result: raise errors.CannotDeleteError('DELETE request failed GET(DELETE(x))==404 invariant') return self.put_document(uri, new_document, check_etag) def delete_element(self, uri, check_etag): d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_delete_element, callbackArgs=(uri, check_etag)) ## Attribute management def _cb_get_attribute(self, response, uri): """This is called when the document that relates to the attribute is retrieved.""" if response.code == 404: raise errors.ResourceNotFound document = response.data xml_doc = etree.parse(StringIO(document)) application = getApplicationForURI(uri) ns_dict = uri.node_selector.get_ns_bindings(application.default_ns) try: xpath = uri.node_selector.replace_default_prefix() attribute = xml_doc.xpath(xpath, namespaces = ns_dict) except Exception, ex: ex.http_error = errors.ResourceNotFound() raise if not attribute: raise errors.ResourceNotFound elif len(attribute) != 1: raise errors.ResourceNotFound('XPATH expression is ambiguous') # TODO # The server MUST NOT add namespace bindings representing namespaces # used by the element or its children, but declared in ancestor elements return StatusResponse(200, response.etag, attribute[0]) def get_attribute(self, uri, check_etag): d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_get_attribute, callbackArgs=(uri, )) def _cb_delete_attribute(self, response, uri, check_etag): if response.code == 404: raise errors.ResourceNotFound document = response.data xml_doc = etree.parse(StringIO(document)) application = getApplicationForURI(uri) ns_dict = uri.node_selector.get_ns_bindings(application.default_ns) try: elem = xml_doc.xpath(uri.node_selector.replace_default_prefix(append_terminal=False),namespaces=ns_dict) except Exception, ex: ex.http_error = errors.ResourceNotFound() raise if not elem: raise errors.ResourceNotFound if len(elem) != 1: raise errors.ResourceNotFound('XPATH expression is ambiguous') elem = elem[0] attribute = uri.node_selector.terminal_selector.attribute if elem.get(attribute): ## check if the attribute exists XXX use KeyError instead del elem.attrib[attribute] else: raise errors.ResourceNotFound new_document = etree.tostring(xml_doc, encoding='UTF-8', xml_declaration=True) return self.put_document(uri, new_document, check_etag) def delete_attribute(self, uri, check_etag): d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_delete_attribute, callbackArgs=(uri, check_etag)) def _cb_put_attribute(self, response, uri, attribute, check_etag): """This is called when the document that relates to the element is retrieved.""" if response.code == 404: raise errors.NoParentError document = response.data xml_doc = etree.parse(StringIO(document)) application = getApplicationForURI(uri) ns_dict = uri.node_selector.get_ns_bindings(application.default_ns) try: elem = xml_doc.xpath(uri.node_selector.replace_default_prefix(append_terminal=False),namespaces=ns_dict) except Exception, ex: ex.http_error = errors.NoParentError() raise if not elem: raise errors.NoParentError if len(elem) != 1: raise errors.NoParentError('XPATH expression is ambiguous') elem = elem[0] attr_name = uri.node_selector.terminal_selector.attribute elem.set(attr_name, attribute) new_document = etree.tostring(xml_doc, encoding='UTF-8', xml_declaration=True) return self.put_document(uri, new_document, check_etag) def put_attribute(self, uri, attribute, check_etag): ## TODO verify if the attribute is valid d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_put_attribute, callbackArgs=(uri, attribute, check_etag)) ## Namespace Bindings def _cb_get_ns_bindings(self, response, uri): """This is called when the document that relates to the element is retrieved.""" if response.code == 404: raise errors.ResourceNotFound document = response.data xml_doc = etree.parse(StringIO(document)) application = getApplicationForURI(uri) ns_dict = uri.node_selector.get_ns_bindings(application.default_ns) try: elem = xml_doc.xpath(uri.node_selector.replace_default_prefix(append_terminal=False),namespaces=ns_dict) except Exception, ex: ex.http_error = errors.ResourceNotFound() raise if not elem: raise errors.ResourceNotFound elif len(elem)!=1: raise errors.ResourceNotFound('XPATH expression is ambiguous') elem = elem[0] namespaces = '' for prefix, ns in elem.nsmap.items(): namespaces += ' xmlns%s="%s"' % (prefix and ':%s' % prefix or '', ns) result = '<%s %s/>' % (elem.tag, namespaces) return StatusResponse(200, response.etag, result) def get_ns_bindings(self, uri, check_etag): d = self.get_document(uri, check_etag) return d.addCallbacks(self._cb_get_ns_bindings, callbackArgs=(uri, )) from xcap.appusage.capabilities import XCAPCapabilitiesApplication from xcap.appusage.dialogrules import DialogRulesApplication from xcap.appusage.directory import XCAPDirectoryApplication from xcap.appusage.icon import IconApplication from xcap.appusage.pidf import PIDFManipulationApplication from xcap.appusage.presrules import PresenceRulesApplication +from xcap.appusage.purge import PurgeApplication from xcap.appusage.resourcelists import ResourceListsApplication from xcap.appusage.rlsservices import RLSServicesApplication from xcap.appusage.test import TestApplication from xcap.appusage.watchers import WatchersApplication storage = ServerConfig.backend.Storage() applications = { DialogRulesApplication.id: DialogRulesApplication(storage), PIDFManipulationApplication.id: PIDFManipulationApplication(storage), PresenceRulesApplication.id: PresenceRulesApplication(storage), PresenceRulesApplication.oma_id: PresenceRulesApplication(storage), + PurgeApplication.id: PurgeApplication(storage), ResourceListsApplication.id: ResourceListsApplication(storage), RLSServicesApplication.id: RLSServicesApplication(storage), TestApplication.id: TestApplication(storage), WatchersApplication.id: WatchersApplication(storage), XCAPCapabilitiesApplication.id: XCAPCapabilitiesApplication(), XCAPDirectoryApplication.id: XCAPDirectoryApplication(storage) } # public GET applications (GET is not challenged for auth) public_get_applications = {IconApplication.id: IconApplication(storage)} applications.update(public_get_applications) for application in ServerConfig.disabled_applications: applications.pop(application, None) namespaces = dict((k, v.default_ns) for (k, v) in applications.items()) def getApplicationForURI(xcap_uri): return applications.get(xcap_uri.application_id, None) __all__ = ['applications', 'namespaces', 'public_get_applications', 'getApplicationForURI', 'ApplicationUsage', 'Backend'] diff --git a/xcap/appusage/purge.py b/xcap/appusage/purge.py new file mode 100644 index 0000000..3604000 --- /dev/null +++ b/xcap/appusage/purge.py @@ -0,0 +1,25 @@ +# Copyright (C) 2011 AG-Projects. +# + +from xcap import errors +from xcap.appusage import ApplicationUsage +from xcap.interfaces.backend import StatusResponse + +class PurgeApplication(ApplicationUsage): + id = "org.openxcap.purge" + default_ns = "http://openxcap.org/ns/purge" + + def _purge_cb(self, result, uri): + return StatusResponse(200) + + def get_document_local(self, uri, check_etag): + d = self.storage.delete_documents(uri) + d.addCallback(self._purge_cb, uri) + return d + + def put_document(self, uri, document, check_etag): + raise errors.ResourceNotFound("This application does not support PUT method") + + def delete_document(self, uri, document, check_etag): + raise errors.ResourceNotFound("This application does not support DELETE method") + diff --git a/xcap/interfaces/backend/database.py b/xcap/interfaces/backend/database.py index 004e15e..f2c5ba7 100644 --- a/xcap/interfaces/backend/database.py +++ b/xcap/interfaces/backend/database.py @@ -1,396 +1,409 @@ # Copyright (C) 2007-2010 AG-Projects. # """Implementation of a database backend.""" import sys from application import log from application.configuration import ConfigSection from application.python.types import Singleton from zope.interface import implements from twisted.cred import credentials, checkers, error as credError from twisted.internet import defer from _mysql_exceptions import IntegrityError import xcap from xcap.interfaces.backend import IStorage, StatusResponse from xcap.dbutil import connectionForURI, repeat_on_error, make_random_etag class Config(ConfigSection): __cfgfile__ = xcap.__cfgfile__ __section__ = 'Database' authentication_db_uri = '' storage_db_uri = '' subscriber_table = 'subscriber' user_col = 'username' domain_col = 'domain' password_col = 'password' ha1_col = 'ha1' xcap_table = 'xcap' if not Config.authentication_db_uri or not Config.storage_db_uri: log.fatal("Authentication DB URI and Storage DB URI must be provided") sys.exit(1) class DBBase(object): def __init__(self): self._db_connect() class PasswordChecker(DBBase): """A credentials checker against a database subscriber table.""" implements(checkers.ICredentialsChecker) credentialInterfaces = (credentials.IUsernamePassword, credentials.IUsernameHashedPassword) def _db_connect(self): self.conn = auth_db_connection(Config.authentication_db_uri) def _query_credentials(self, credentials): raise NotImplementedError def _got_query_results(self, rows, credentials): if not rows: raise credError.UnauthorizedLogin("Unauthorized login") else: return self._authenticate_credentials(rows[0][0], credentials) def _authenticate_credentials(self, password, credentials): raise NotImplementedError def _checkedPassword(self, matched, username, realm): if matched: username = username.split('@', 1)[0] ## this is the avatar ID return "%s@%s" % (username, realm) else: raise credError.UnauthorizedLogin("Unauthorized login") def requestAvatarId(self, credentials): """Return the avatar ID for the credentials which must have the username and realm attributes, or an UnauthorizedLogin in case of a failure.""" d = self._query_credentials(credentials) return d class PlainPasswordChecker(PasswordChecker): """A credentials checker against a database subscriber table, where the passwords are stored in plain text.""" implements(checkers.ICredentialsChecker) def _query_credentials(self, credentials): username, domain = credentials.username.split('@', 1)[0], credentials.realm query = """SELECT %(password_col)s FROM %(table)s WHERE %(user_col)s = %%(username)s AND %(domain_col)s = %%(domain)s""" % { "password_col": Config.password_col, "user_col": Config.user_col, "domain_col": Config.domain_col, "table": Config.subscriber_table } params = {"username": username, "domain": domain} return self.conn.runQuery(query, params).addCallback(self._got_query_results, credentials) def _authenticate_credentials(self, hash, credentials): return defer.maybeDeferred( credentials.checkPassword, hash).addCallback( self._checkedPassword, credentials.username, credentials.realm) class HashPasswordChecker(PasswordChecker): """A credentials checker against a database subscriber table, where the passwords are stored as MD5 hashes.""" implements(checkers.ICredentialsChecker) def _query_credentials(self, credentials): username, domain = credentials.username.split('@', 1)[0], credentials.realm query = """SELECT %(ha1_col)s FROM %(table)s WHERE %(user_col)s = %%(username)s AND %(domain_col)s = %%(domain)s""" % { "ha1_col": Config.ha1_col, "user_col": Config.user_col, "domain_col": Config.domain_col, "table": Config.subscriber_table} params = {"username": username, "domain": domain} return self.conn.runQuery(query, params).addCallback(self._got_query_results, credentials) def _authenticate_credentials(self, hash, credentials): return defer.maybeDeferred( credentials.checkHash, hash).addCallback( self._checkedPassword, credentials.username, credentials.realm) class Error(Exception): def __init__(self): if hasattr(self, 'msg'): return Exception.__init__(self, self.msg) else: return Exception.__init__(self) class RaceError(Error): """The errors of this type are raised for the requests that failed because of concurrent modification of the database by other clients. For example, before DELETE we do SELECT first, to check that a document of the right etag exists. The actual check is performed by a function in twisted that is passed as a callback. Then etag from the SELECT request is used in the DELETE request. This seems unnecessary convoluted and probably should be changed to 'DELETE .. WHERE etag=ETAG'. We still need to find out whether DELETE was actually performed. """ class UpdateFailed(RaceError): msg = 'UPDATE request failed' class DeleteFailed(RaceError): msg = 'DELETE request failed' class MultipleResultsError(Error): """This should never happen. If it did happen. that means either the table was corrupted or there's a logic error""" def __init__(self, params): Exception.__init__(self, 'database request has more than one result: ' + repr(params)) class Storage(DBBase): __metaclass__ = Singleton implements(IStorage) app_mapping = {"pres-rules" : 1<<1, "org.openmobilealliance.pres-rules" : 1<<1, "resource-lists" : 1<<2, "rls-services" : 1<<3, "pidf-manipulation" : 1<<4, "org.openxcap.dialog-rules" : 1<<5, "oma_status-icon" : 1<<6, "test-app" : 0} def _db_connect(self): self.conn = storage_db_connection(Config.storage_db_uri) def _normalize_document_path(self, uri): if uri.application_id in ("pres-rules", "org.openmobilealliance.pres-rules"): # some clients e.g. counterpath's eyebeam save presence rules under # different filenames between versions and they expect to find the same # information, thus we are forcing all presence rules documents to be # saved under "index.xml" default filename uri.doc_selector.document_path = "index.xml" def _get_document(self, trans, uri, check_etag): username, domain = uri.user.username, uri.user.domain self._normalize_document_path(uri) doc_type = self.app_mapping[uri.application_id] query = """SELECT doc, etag FROM %(table)s WHERE username = %%(username)s AND domain = %%(domain)s AND doc_type= %%(doc_type)s AND doc_uri=%%(document_path)s""" % { "table": Config.xcap_table} params = {"username": username, "domain" : domain, "doc_type": doc_type, "document_path": uri.doc_selector.document_path} trans.execute(query, params) result = trans.fetchall() if len(result)>1: raise MultipleResultsError(params) elif result: doc, etag = result[0] if isinstance(doc, unicode): doc = doc.encode('utf-8') check_etag(etag) response = StatusResponse(200, etag, doc) else: response = StatusResponse(404) return response def _put_document(self, trans, uri, document, check_etag): username, domain = uri.user.username, uri.user.domain self._normalize_document_path(uri) doc_type = self.app_mapping[uri.application_id] document_path = uri.doc_selector.document_path query = """SELECT etag FROM %(table)s WHERE username = %%(username)s AND domain = %%(domain)s AND doc_type= %%(doc_type)s AND doc_uri=%%(document_path)s""" % { "table": Config.xcap_table} params = {"username": username, "domain" : domain, "doc_type": doc_type, "document_path": document_path} trans.execute(query, params) result = trans.fetchall() if len(result) > 1: raise MultipleResultsError(params) elif not result: check_etag(None, False) ## the document doesn't exist, create it etag = make_random_etag(uri) query = """INSERT INTO %(table)s (username, domain, doc_type, etag, doc, doc_uri) VALUES (%%(username)s, %%(domain)s, %%(doc_type)s, %%(etag)s, %%(document)s, %%(document_path)s)""" % { "table": Config.xcap_table } params = {"username": username, "domain" : domain, "doc_type": doc_type, "etag": etag, "document": document, "document_path": document_path} # may raise IntegrityError here, if the document was created in another connection # will be catched by repeat_on_error trans.execute(query, params) return StatusResponse(201, etag) else: old_etag = result[0][0] ## first check the etag of the existing resource check_etag(old_etag) ## the document exists, replace it etag = make_random_etag(uri) query = """UPDATE %(table)s SET doc = %%(document)s, etag = %%(etag)s WHERE username = %%(username)s AND domain = %%(domain)s AND doc_type = %%(doc_type)s AND etag = %%(old_etag)s AND doc_uri = %%(document_path)s""" % { "table": Config.xcap_table } params = {"document": document, "etag": etag, "username": username, "domain" : domain, "doc_type": doc_type, "old_etag": old_etag, "document_path": document_path} trans.execute(query, params) # the request may not update anything (e.g. if etag was changed by another connection # after we did SELECT); if so, we should retry updated = getattr(trans._connection, 'affected_rows', lambda : 1)() if not updated: raise UpdateFailed assert updated == 1, updated return StatusResponse(200, etag, old_etag=old_etag) def _delete_document(self, trans, uri, check_etag): username, domain = uri.user.username, uri.user.domain self._normalize_document_path(uri) doc_type = self.app_mapping[uri.application_id] document_path = uri.doc_selector.document_path query = """SELECT etag FROM %(table)s WHERE username = %%(username)s AND domain = %%(domain)s AND doc_type= %%(doc_type)s AND doc_uri = %%(document_path)s""" % { "table": Config.xcap_table} params = {"username": username, "domain" : domain, "doc_type": doc_type, "document_path": document_path} trans.execute(query, params) result = trans.fetchall() if len(result)>1: raise MultipleResultsError(params) elif result: etag = result[0][0] check_etag(etag) query = """DELETE FROM %(table)s WHERE username = %%(username)s AND domain = %%(domain)s AND doc_type= %%(doc_type)s AND doc_uri = %%(document_path)s AND etag = %%(etag)s""" % {"table" : Config.xcap_table} params = {"username": username, "domain" : domain, "doc_type": doc_type, "document_path": document_path, "etag": etag} trans.execute(query, params) deleted = getattr(trans._connection, 'affected_rows', lambda : 1)() if not deleted: # the document was replaced/removed after the SELECT but before the DELETE raise DeleteFailed assert deleted == 1, deleted return StatusResponse(200, old_etag=etag) else: return StatusResponse(404) + def _delete_all_documents(self, trans, uri): + username, domain = uri.user.username, uri.user.domain + query = """DELETE FROM %(table)s + WHERE username = %%(username)s AND domain = %%(domain)s + """ % {"table" : Config.xcap_table} + params = {"username": username, + "domain" : domain} + trans.execute(query, params) + return StatusResponse(200) + def get_document(self, uri, check_etag): return self.conn.runInteraction(self._get_document, uri, check_etag) def put_document(self, uri, document, check_etag): return repeat_on_error(10, (UpdateFailed, IntegrityError), self.conn.runInteraction, self._put_document, uri, document, check_etag) def delete_document(self, uri, check_etag): return repeat_on_error(10, DeleteFailed, self.conn.runInteraction, self._delete_document, uri, check_etag) + def delete_documents(self, uri, check_etag): + return self.conn.runInteraction(self._delete_all_documents, uri) + # Application-specific functions def _get_watchers(self, trans, uri): status_mapping = {1: "allow", 2: "confirm", 3: "deny"} presentity_uri = "sip:%s@%s" % (uri.user.username, uri.user.domain) query = """SELECT watcher_username, watcher_domain, status FROM watchers WHERE presentity_uri = %(puri)s""" params = {'puri': presentity_uri} trans.execute(query, params) result = trans.fetchall() watchers = [{"id": "%s@%s" % (w_user, w_domain), "status": status_mapping.get(subs_status, "unknown"), "online": "false"} for w_user, w_domain, subs_status in result] query = """SELECT watcher_username, watcher_domain FROM active_watchers WHERE presentity_uri = %(puri)s AND event = 'presence'""" trans.execute(query, params) result = trans.fetchall() active_watchers = set("%s@%s" % pair for pair in result) for watcher in watchers: if watcher["id"] in active_watchers: watcher["online"] = "true" return watchers def get_watchers(self, uri): return self.conn.runInteraction(self._get_watchers, uri) def _get_documents_list(self, trans, uri): query = """SELECT doc_type, doc_uri, etag FROM %(table)s WHERE username = %%(username)s AND domain = %%(domain)s""" % {'table': Config.xcap_table} params = {'username': uri.user.username, 'domain': uri.user.domain} trans.execute(query, params) result = trans.fetchall() docs = {} for r in result: app = [k for k, v in self.app_mapping.iteritems() if v == r[0]][0] if docs.has_key(app): docs[app].append((r[1], r[2])) else: docs[app] = [(r[1], r[2])] # Ex: {'pres-rules': [('index.html', '4564fd9c9a2a2e3e796310b00c9908aa')]} return docs def get_documents_list(self, uri): return self.conn.runInteraction(self._get_documents_list, uri) installSignalHandlers = True def auth_db_connection(uri): conn = connectionForURI(uri) return conn def storage_db_connection(uri): conn = connectionForURI(uri) def cb(res): if res[0:1][0:1] and res[0][0]: print '%s xcap documents in the database' % res[0][0] return res def eb(fail): fail.printTraceback() return fail # connect early, so database problem are detected early d = conn.runQuery('SELECT count(*) from %s' % Config.xcap_table) d.addCallback(cb) d.addErrback(eb) return conn diff --git a/xcap/interfaces/backend/sipthor.py b/xcap/interfaces/backend/sipthor.py index d677fba..3b8f7b7 100644 --- a/xcap/interfaces/backend/sipthor.py +++ b/xcap/interfaces/backend/sipthor.py @@ -1,582 +1,603 @@ # Copyright (C) 2007-2010 AG-Projects. # # This module is proprietary to AG Projects. Use of this module by third # parties is not supported. import re import signal import cjson from formencode import validators from application import log from application.notification import IObserver, NotificationCenter from application.python import Null from application.python.types import Singleton from application.system import host from application.process import process from application.configuration import ConfigSection, ConfigSetting from sqlobject import sqlhub, connectionForURI, SQLObject, AND from sqlobject import StringCol, IntCol, DateTimeCol, SOBLOBCol, Col from sqlobject import MultipleJoin, ForeignKey from zope.interface import implements from twisted.internet import reactor from twisted.internet import defer from twisted.internet.defer import Deferred, maybeDeferred from twisted.cred.checkers import ICredentialsChecker from twisted.cred.credentials import IUsernamePassword, IUsernameHashedPassword from twisted.cred.error import UnauthorizedLogin from thor.link import ControlLink, Notification, Request from thor.eventservice import EventServiceClient, ThorEvent from thor.entities import ThorEntitiesRoleMap, GenericThorEntity as ThorEntity from gnutls.interfaces.twisted import X509Credentials from gnutls.constants import COMP_DEFLATE, COMP_LZO, COMP_NULL from sipsimple.core import Engine, FromHeader, Header, Publication, RouteHeader, SIPURI from sipsimple.threading import run_in_twisted_thread import xcap from xcap.tls import Certificate, PrivateKey from xcap.interfaces.backend import StatusResponse from xcap.datatypes import XCAPRootURI from xcap.dbutil import make_random_etag from xcap.xcapdiff import Notifier class ThorNodeConfig(ConfigSection): __cfgfile__ = xcap.__cfgfile__ __section__ = 'ThorNetwork' domain = "sipthor.net" multiply = 1000 certificate = ConfigSetting(type=Certificate, value=None) private_key = ConfigSetting(type=PrivateKey, value=None) ca = ConfigSetting(type=Certificate, value=None) class JSONValidator(validators.Validator): def to_python(self, value, state): if value is None: return None try: return cjson.decode(value) except Exception: raise validators.Invalid("expected a decodable JSON object in the JSONCol '%s', got %s %r instead" % (self.name, type(value), value), value, state) def from_python(self, value, state): if value is None: return None try: return cjson.encode(value) except Exception: raise validators.Invalid("expected an encodable JSON object in the JSONCol '%s', got %s %r instead" % (self.name, type(value), value), value, state) class SOJSONCol(SOBLOBCol): def createValidators(self): return [JSONValidator()] + super(SOJSONCol, self).createValidators() class JSONCol(Col): baseClass = SOJSONCol class SipAccount(SQLObject): class sqlmeta: table = 'sip_accounts_meta' username = StringCol(length=64) domain = StringCol(length=64) firstName = StringCol(length=64) lastName = StringCol(length=64) email = StringCol(length=64) customerId = IntCol(default=0) resellerId = IntCol(default=0) ownerId = IntCol(default=0) changeDate = DateTimeCol(default=DateTimeCol.now) ## joins data = MultipleJoin('SipAccountData', joinColumn='account_id') def _set_profile(self, value): data = list(self.data) if not data: SipAccountData(account=self, profile=value) else: data[0].profile = value def _get_profile(self): return self.data[0].profile def set(self, **kwargs): kwargs = kwargs.copy() profile = kwargs.pop('profile', None) SQLObject.set(self, **kwargs) if profile is not None: self._set_profile(profile) class SipAccountData(SQLObject): class sqlmeta: table = 'sip_accounts_data' account = ForeignKey('SipAccount', cascade=True) profile = JSONCol() def sanitize_application_id(application_id): if application_id == "org.openmobilealliance.pres-rules": return "pres-rules" else: return application_id class ThorEntityAddress(str): def __new__(cls, ip, control_port=None, version='unknown'): instance = str.__new__(cls, ip) instance.ip = ip instance.version = version instance.control_port = control_port return instance class GetSIPWatchers(Request): def __new__(cls, account): command = "get sip_watchers for %s" % account instance = Request.__new__(cls, command) return instance class XCAPProvisioning(EventServiceClient): __metaclass__ = Singleton topics = ["Thor.Members"] def __init__(self): self._database = DatabaseConnection() self.node = ThorEntity(host.default_ip, ['xcap_server'], version=xcap.__version__) self.networks = {} self.presence_message = ThorEvent('Thor.Presence', self.node.id) self.shutdown_message = ThorEvent('Thor.Leave', self.node.id) credentials = X509Credentials(ThorNodeConfig.certificate, ThorNodeConfig.private_key, [ThorNodeConfig.ca]) credentials.verify_peer = True credentials.session_params.compressions = (COMP_LZO, COMP_DEFLATE, COMP_NULL) self.control = ControlLink(credentials) EventServiceClient.__init__(self, ThorNodeConfig.domain, credentials) process.signals.add_handler(signal.SIGHUP, self._handle_SIGHUP) process.signals.add_handler(signal.SIGINT, self._handle_SIGINT) process.signals.add_handler(signal.SIGTERM, self._handle_SIGTERM) def _disconnect_all(self, result): self.control.disconnect_all() EventServiceClient._disconnect_all(self, result) def lookup(self, key): network = self.networks.get("sip_proxy", None) if network is None: return None try: node = network.lookup_node(key) except LookupError: node = None except Exception: log.err() node = None return node def notify(self, operation, entity_type, entity): node = self.lookup(entity) if node is not None: if node.control_port is None: log.error("Could not send notify because node %s has no control port" % node.ip) return self.control.send_request(Notification("notify %s %s %s" % (operation, entity_type, entity)), (node.ip, node.control_port)) def get_watchers(self, key): node = self.lookup(key) if node is None: return defer.fail("no nodes found when searching for key %s" % str(key)) if node.control_port is None: return defer.fail("could not send notify because node %s has no control port" % node.ip) request = GetSIPWatchers(key) request.deferred = Deferred() self.control.send_request(request, (node.ip, node.control_port)) return request.deferred def handle_event(self, event): # print "Received event: %s" % event networks = self.networks role_map = ThorEntitiesRoleMap(event.message) ## mapping between role names and lists of nodes with that role thor_databases = role_map.get('thor_database', []) if thor_databases: thor_databases.sort(lambda x, y: cmp(x.priority, y.priority) or cmp(x.ip, y.ip)) dburi = thor_databases[0].dburi else: dburi = None self._database.update_dburi(dburi) all_roles = role_map.keys() + networks.keys() for role in all_roles: try: network = networks[role] ## avoid setdefault here because it always evaluates the 2nd argument except KeyError: from thor import network as thor_network if role in ["thor_manager", "thor_monitor", "provisioning_server", "media_relay", "thor_database"]: continue else: network = thor_network.new(ThorNodeConfig.multiply) networks[role] = network new_nodes = set([ThorEntityAddress(node.ip, getattr(node, 'control_port', None), getattr(node, 'version', 'unknown')) for node in role_map.get(role, [])]) old_nodes = set(network.nodes) ## compute set differences added_nodes = new_nodes - old_nodes removed_nodes = old_nodes - new_nodes if removed_nodes: for node in removed_nodes: network.remove_node(node) self.control.discard_link((node.ip, node.control_port)) plural = len(removed_nodes) != 1 and 's' or '' log.msg("removed %s node%s: %s" % (role, plural, ', '.join(removed_nodes))) if added_nodes: for node in added_nodes: network.add_node(node) plural = len(added_nodes) != 1 and 's' or '' log.msg("added %s node%s: %s" % (role, plural, ', '.join(added_nodes))) #print "Thor %s nodes: %s" % (role, str(network.nodes)) class NotFound(Exception): pass class NoDatabase(Exception): pass class DatabaseConnection(object): __metaclass__ = Singleton def __init__(self): self.dburi = None # Methods to be called from the Twisted thread: def put(self, uri, document, check_etag, new_etag): defer = Deferred() operation = lambda profile: self._put_operation(uri, document, check_etag, new_etag, profile) reactor.callInThread(self.retrieve_profile, uri.user.username, uri.user.domain, operation, True, defer) return defer def delete(self, uri, check_etag): defer = Deferred() operation = lambda profile: self._delete_operation(uri, check_etag, profile) reactor.callInThread(self.retrieve_profile, uri.user.username, uri.user.domain, operation, True, defer) return defer + def delete_all(self, uri): + defer = Deferred() + operation = lambda profile: self._delete_all_operation(uri, profile) + reactor.callInThread(self.retrieve_profile, uri.user.username, uri.user.domain, operation, True, defer) + return defer + def get(self, uri): defer = Deferred() operation = lambda profile: self._get_operation(uri, profile) reactor.callInThread(self.retrieve_profile, uri.user.username, uri.user.domain, operation, False, defer) return defer def get_profile(self, username, domain): defer = Deferred() reactor.callInThread(self.retrieve_profile, username, domain, lambda profile: profile, False, defer) return defer def get_documents_list(self, uri): defer = Deferred() operation = lambda profile: self._get_documents_list_operation(uri, profile) reactor.callInThread(self.retrieve_profile, uri.user.username, uri.user.domain, operation, False, defer) return defer # Methods to be called in a separate thread: def _put_operation(self, uri, document, check_etag, new_etag, profile): application_id = sanitize_application_id(uri.application_id) xcap_docs = profile.setdefault("xcap", {}) try: etag = xcap_docs[application_id][uri.doc_selector.document_path][1] except KeyError: found = False etag = None check_etag(None, False) else: found = True check_etag(etag) xcap_app = xcap_docs.setdefault(application_id, {}) xcap_app[uri.doc_selector.document_path] = (document, new_etag) return (found, etag, new_etag) def _delete_operation(self, uri, check_etag, profile): application_id = sanitize_application_id(uri.application_id) xcap_docs = profile.setdefault("xcap", {}) try: etag = xcap_docs[application_id][uri.doc_selector.document_path][1] except KeyError: raise NotFound() check_etag(etag) del(xcap_docs[application_id][uri.doc_selector.document_path]) return (etag) + def _delete_all_operation(self, uri, profile): + xcap_docs = profile.setdefault("xcap", {}) + xcap_docs.clear() + return None + def _get_operation(self, uri, profile): try: xcap_docs = profile["xcap"] doc, etag = xcap_docs[sanitize_application_id(uri.application_id)][uri.doc_selector.document_path] except KeyError: raise NotFound() return doc, etag def _get_documents_list_operation(self, uri, profile): try: xcap_docs = profile["xcap"] except KeyError: raise NotFound() return xcap_docs def retrieve_profile(self, username, domain, operation, update, defer): transaction = None try: if self.dburi is None: raise NoDatabase() transaction = sqlhub.processConnection.transaction() try: db_account = SipAccount.select(AND(SipAccount.q.username == username, SipAccount.q.domain == domain), connection = transaction, forUpdate = update)[0] except IndexError: raise NotFound() profile = db_account.profile result = operation(profile) # NB: may modify profile! if update: db_account.profile = profile transaction.commit(close=True) except Exception, e: if transaction: transaction.rollback() reactor.callFromThread(defer.errback, e) else: reactor.callFromThread(defer.callback, result) def update_dburi(self, dburi): if self.dburi != dburi: if self.dburi is not None: sqlhub.processConnection.close() if dburi is None: sqlhub.processConnection else: sqlhub.processConnection = connectionForURI(dburi) self.dburi = dburi class SipthorPasswordChecker(object): implements(ICredentialsChecker) credentialInterfaces = (IUsernamePassword, IUsernameHashedPassword) def __init__(self): self._database = DatabaseConnection() def _query_credentials(self, credentials): username, domain = credentials.username.split('@', 1)[0], credentials.realm result = self._database.get_profile(username, domain) result.addCallback(self._got_query_results, credentials) result.addErrback(self._got_unsuccessfull) return result def _got_unsuccessfull(self, failure): failure.trap(NotFound) raise UnauthorizedLogin("Unauthorized login") def _got_query_results(self, profile, credentials): return self._authenticate_credentials(profile, credentials) def _authenticate_credentials(self, profile, credentials): raise NotImplementedError def _checkedPassword(self, matched, username, realm): if matched: username = username.split('@', 1)[0] ## this is the avatar ID return "%s@%s" % (username, realm) else: raise UnauthorizedLogin("Unauthorized login") def requestAvatarId(self, credentials): """Return the avatar ID for the credentials which must have the username and realm attributes, or an UnauthorizedLogin in case of a failure.""" d = self._query_credentials(credentials) return d class PlainPasswordChecker(SipthorPasswordChecker): """A credentials checker against a database subscriber table, where the passwords are stored in plain text.""" implements(ICredentialsChecker) def _authenticate_credentials(self, profile, credentials): return maybeDeferred( credentials.checkPassword, profile["password"]).addCallback( self._checkedPassword, credentials.username, credentials.realm) class HashPasswordChecker(SipthorPasswordChecker): """A credentials checker against a database subscriber table, where the passwords are stored as MD5 hashes.""" implements(ICredentialsChecker) def _authenticate_credentials(self, profile, credentials): return maybeDeferred( credentials.checkHash, profile["ha1"]).addCallback( self._checkedPassword, credentials.username, credentials.realm) class ServerConfig(ConfigSection): __cfgfile__ = xcap.__cfgfile__ __section__ = 'Server' root = ConfigSetting(type=XCAPRootURI, value=None) class SIPNotifier(object): __metaclass__ = Singleton implements(IObserver) def __init__(self): self.provisioning = XCAPProvisioning() self.engine = Engine() self.engine.start( user_agent="OpenXCAP %s" % xcap.__version__, ) def send_publish(self, uri, body): uri = re.sub("^(sip:|sips:)", "", uri) destination_node = self.provisioning.lookup(uri) if destination_node is not None: # TODO: add configuration settings for SIP transport. -Saul publication = Publication(FromHeader(SIPURI(uri)), "xcap-diff", "application/xcap-diff+xml", duration=0, extra_headers=[Header('Thor-Scope', 'publish-xcap')]) NotificationCenter().add_observer(self, sender=publication) route_header = RouteHeader(SIPURI(host=str(destination_node), port='5060', parameters=dict(transport='udp'))) publication.publish(body, route_header, timeout=5) @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPPublicationDidSucceed(self, notification): log.msg("PUBLISH for xcap-diff event successfully sent to %s for %s" % (notification.data.route_header.uri, notification.sender.from_header.uri)) def _NH_SIPPublicationDidEnd(self, notification): log.msg("PUBLISH for xcap-diff event ended for %s" % notification.sender.from_header.uri) NotificationCenter().remove_observer(self, sender=notification.sender) def _NH_SIPPublicationDidFail(self, notification): log.msg("PUBLISH for xcap-diff event failed to %s for %s" % (notification.data.route_header.uri, notification.sender.from_header.uri)) NotificationCenter().remove_observer(self, sender=notification.sender) class Storage(object): __metaclass__ = Singleton def __init__(self): self._database = DatabaseConnection() self._provisioning = XCAPProvisioning() self._sip_notifier = SIPNotifier() self._notifier = Notifier(ServerConfig.root, self._sip_notifier.send_publish) def _normalize_document_path(self, uri): if uri.application_id in ("pres-rules", "org.openmobilealliance.pres-rules"): # some clients e.g. counterpath's eyebeam save presence rules under # different filenames between versions and they expect to find the same # information, thus we are forcing all presence rules documents to be # saved under "index.xml" default filename uri.doc_selector.document_path = "index.xml" def get_document(self, uri, check_etag): self._normalize_document_path(uri) result = self._database.get(uri) result.addCallback(self._got_document, check_etag) result.addErrback(self._eb_not_found) return result def _eb_not_found(self, failure): failure.trap(NotFound) return StatusResponse(404) def _got_document(self, (doc, etag), check_etag): check_etag(etag) return StatusResponse(200, etag, doc) def put_document(self, uri, document, check_etag): self._normalize_document_path(uri) etag = make_random_etag(uri) result = self._database.put(uri, document, check_etag, etag) result.addCallback(self._cb_put, uri, "%s@%s" % (uri.user.username, uri.user.domain)) result.addErrback(self._eb_not_found) return result def _cb_put(self, result, uri, thor_key): if result[0]: code = 200 else: code = 201 self._provisioning.notify("update", "sip_account", thor_key) self._notifier.on_change(uri, result[1], result[2]) return StatusResponse(code, result[2]) + def delete_documents(self, uri): + result = self._database.delete_all(uri) + result.addCallback(self._cb_delete_all, uri, "%s@%s" % (uri.user.username, uri.user.domain)) + result.addErrback(self._eb_not_found) + return result + + def _cb_delete_all(self, result, uri, thor_key): + self._provisioning.notify("update", "sip_account", thor_key) + return StatusResponse(200) + def delete_document(self, uri, check_etag): self._normalize_document_path(uri) result = self._database.delete(uri, check_etag) result.addCallback(self._cb_delete, uri, "%s@%s" % (uri.user.username, uri.user.domain)) result.addErrback(self._eb_not_found) return result def _cb_delete(self, result, uri, thor_key): self._provisioning.notify("update", "sip_account", thor_key) self._notifier.on_change(uri, result[1], None) return StatusResponse(200) def get_watchers(self, uri): thor_key = "%s@%s" % (uri.user.username, uri.user.domain) result = self._provisioning.get_watchers(thor_key) result.addCallback(self._get_watchers_decode) return result def _get_watchers_decode(self, response): if response.code == 200: watchers = cjson.decode(response.data) for watcher in watchers: watcher["online"] = str(watcher["online"]).lower() return watchers else: print "error: %s" % response def get_documents_list(self, uri): result = self._database.get_documents_list(uri) result.addCallback(self._got_documents_list) result.addErrback(self._got_documents_list_error) return result def _got_documents_list(self, xcap_docs): docs = {} if xcap_docs: for k, v in xcap_docs.iteritems(): for k2, v2 in v.iteritems(): if docs.has_key(k): docs[k].append((k2, v2[1])) else: docs[k] = [(k2, v2[1])] return docs def _got_documents_list_error(self, failure): failure.trap(NotFound) return {} installSignalHandlers = False diff --git a/xcap/resource.py b/xcap/resource.py index feb975a..8d97c77 100644 --- a/xcap/resource.py +++ b/xcap/resource.py @@ -1,128 +1,131 @@ # Copyright (C) 2007-2010 AG Projects. # """XCAP resources module""" from twisted.web2 import http, resource, responsecode from twisted.web2.http_headers import ETag, MimeType from twisted.web2.static import MetaDataMixin class XCAPResource(resource.Resource, resource.LeafResource, MetaDataMixin): def __init__(self, xcap_uri, application): self.xcap_uri = xcap_uri self.application = application self.e_tag = None def checkPreconditions(self, request): ## don't let renderHTTP to automatically check preconditions, we'll do this ourselves return None def checkEtag(self, request, etag, exists=True): http.checkPreconditions(request, etag=ETag(etag), entityExists=exists) def renderHTTP(self, request): d = resource.Resource.renderHTTP(self, request) d.addCallback(self.sendResponse) return d def setHeaders(self, response): ## Don't provide additional resource information to error responses, ## this is already done by the responses in the errors module if response.code < 400: for (header, value) in ( ("etag", self.etag()), ("content-type", self.contentType()) ): if value is not None: response.headers.setHeader(header, value) return response def sendResponse(self, response): if response.etag: self.e_tag = ETag(response.etag) response = http.Response(response.code, stream=response.data) return self.setHeaders(response) def etag(self): return self.e_tag or None class XCAPDocument(XCAPResource): def http_GET(self, request): d = self.application.get_document(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d def http_PUT(self, request): application = self.application document = request.attachment return application.put_document(self.xcap_uri, document, lambda e, exists=True: self.checkEtag(request, e, exists)) def http_DELETE(self, request): d = self.application.delete_document(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d def contentType(self): - return MimeType.fromString(self.application.mime_type) + try: + return MimeType.fromString(self.application.mime_type) + except TypeError: + return None class XCAPElement(XCAPResource): content_type = MimeType.fromString("application/xcap-el+xml") def http_GET(self, request): d = self.application.get_element(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d def http_DELETE(self, request): d = self.application.delete_element(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d def http_PUT(self, request): content_type = request.headers.getHeader('content-type') if not content_type or content_type != self.content_type: raise http.HTTPError(responsecode.UNSUPPORTED_MEDIA_TYPE) element = request.attachment d = self.application.put_element(self.xcap_uri, element, lambda e: self.checkEtag(request, e)) return d def contentType(self): return self.content_type class XCAPAttribute(XCAPResource): content_type = MimeType.fromString("application/xcap-att+xml") def contentType(self): return self.content_type def http_GET(self, request): d = self.application.get_attribute(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d def http_DELETE(self, request): d = self.application.delete_attribute(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d def http_PUT(self, request): content_type = request.headers.getHeader('content-type') if not content_type or content_type != self.content_type: raise http.HTTPError(responsecode.UNSUPPORTED_MEDIA_TYPE) attribute = request.attachment d = self.application.put_attribute(self.xcap_uri, attribute, lambda e: self.checkEtag(request, e)) return d class XCAPNamespaceBinding(XCAPResource): content_type = MimeType.fromString("application/xcap-ns+xml") def contentType(self): return self.content_type def http_GET(self, request): d = self.application.get_ns_bindings(self.xcap_uri, lambda e: self.checkEtag(request, e)) return d