This file is indexed.

/usr/share/pyshared/pyxmpp/error.py is in python-pyxmpp 1.1.2-1.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
#
# (C) Copyright 2003-2010 Jacek Konieczny <jajcus@jajcus.net>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License Version
# 2.1 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#

"""XMPP error handling.

Normative reference:
  - `RFC 3920 <http://www.ietf.org/rfc/rfc3920.txt>`__
  - `JEP 86 <http://www.jabber.org/jeps/jep-0086.html>`__
"""

__docformat__="restructuredtext en"

import libxml2

from pyxmpp.utils import from_utf8, to_utf8
from pyxmpp.xmlextra import common_doc, common_root, common_ns
from pyxmpp import xmlextra
from pyxmpp.exceptions import ProtocolError

stream_errors={
            u"bad-format":
                ("Received XML cannot be processed",),
            u"bad-namespace-prefix":
                ("Bad namespace prefix",),
            u"conflict":
                ("Closing stream because of conflicting stream being opened",),
            u"connection-timeout":
                ("Connection was idle too long",),
            u"host-gone":
                ("Hostname is no longer hosted on the server",),
            u"host-unknown":
                ("Hostname requested is not known to the server",),
            u"improper-addressing":
                ("Improper addressing",),
            u"internal-server-error":
                ("Internal server error",),
            u"invalid-from":
                ("Invalid sender address",),
            u"invalid-id":
                ("Invalid stream ID",),
            u"invalid-namespace":
                ("Invalid namespace",),
            u"invalid-xml":
                ("Invalid XML",),
            u"not-authorized":
                ("Not authorized",),
            u"policy-violation":
                ("Local policy violation",),
            u"remote-connection-failed":
                ("Remote connection failed",),
            u"resource-constraint":
                ("Remote connection failed",),
            u"restricted-xml":
                ("Restricted XML received",),
            u"see-other-host":
                ("Redirection required",),
            u"system-shutdown":
                ("The server is being shut down",),
            u"undefined-condition":
                ("Unknown error",),
            u"unsupported-encoding":
                ("Unsupported encoding",),
            u"unsupported-stanza-type":
                ("Unsupported stanza type",),
            u"unsupported-version":
                ("Unsupported protocol version",),
            u"xml-not-well-formed":
                ("XML sent by client is not well formed",),
    }

stanza_errors={
            u"bad-request":
                ("Bad request",
                "modify",400),
            u"conflict":
                ("Named session or resource already exists",
                "cancel",409),
            u"feature-not-implemented":
                ("Feature requested is not implemented",
                "cancel",501),
            u"forbidden":
                ("You are forbidden to perform requested action",
                "auth",403),
            u"gone":
                ("Recipient or server can no longer be contacted at this address",
                "modify",302),
            u"internal-server-error":
                ("Internal server error",
                "wait",500),
            u"item-not-found":
                ("Item not found"
                ,"cancel",404),
            u"jid-malformed":
                ("JID malformed",
                "modify",400),
            u"not-acceptable":
                ("Requested action is not acceptable",
                "modify",406),
            u"not-allowed":
                ("Requested action is not allowed",
                "cancel",405),
            u"not-authorized":
                ("Not authorized",
                "auth",401),
            u"payment-required":
                ("Payment required",
                "auth",402),
            u"recipient-unavailable":
                ("Recipient is not available",
                "wait",404),
            u"redirect":
                ("Redirection",
                "modify",302),
            u"registration-required":
                ("Registration required",
                "auth",407),
            u"remote-server-not-found":
                ("Remote server not found",
                "cancel",404),
            u"remote-server-timeout":
                ("Remote server timeout",
                "wait",504),
            u"resource-constraint":
                ("Resource constraint",
                "wait",500),
            u"service-unavailable":
                ("Service is not available",
                "cancel",503),
            u"subscription-required":
                ("Subscription is required",
                "auth",407),
            u"undefined-condition":
                ("Unknown error",
                "cancel",500),
            u"unexpected-request":
                ("Unexpected request",
                "wait",400),
    }

legacy_codes={
        302: "redirect",
        400: "bad-request",
        401: "not-authorized",
        402: "payment-required",
        403: "forbidden",
        404: "item-not-found",
        405: "not-allowed",
        406: "not-acceptable",
        407: "registration-required",
        408: "remote-server-timeout",
        409: "conflict",
        500: "internal-server-error",
        501: "feature-not-implemented",
        502: "service-unavailable",
        503: "service-unavailable",
        504: "remote-server-timeout",
        510: "service-unavailable",
    }

STANZA_ERROR_NS='urn:ietf:params:xml:ns:xmpp-stanzas'
STREAM_ERROR_NS='urn:ietf:params:xml:ns:xmpp-streams'
PYXMPP_ERROR_NS='http://pyxmpp.jajcus.net/xmlns/errors'
STREAM_NS="http://etherx.jabber.org/streams"

class ErrorNode:
    """Base class for both XMPP stream and stanza errors"""
    def __init__(self,xmlnode_or_cond,ns=None,copy=True,parent=None):
        """Initialize an ErrorNode object.

        :Parameters:
            - `xmlnode_or_cond`: XML node to be wrapped into this object
              or error condition name.
            - `ns`: XML namespace URI of the error condition element (to be
              used when the provided node has no namespace).
            - `copy`: When `True` then the XML node will be copied,
              otherwise it is only borrowed.
            - `parent`: Parent node for the XML node to be copied or created.
        :Types:
            - `xmlnode_or_cond`: `libxml2.xmlNode` or `unicode`
            - `ns`: `unicode`
            - `copy`: `bool`
            - `parent`: `libxml2.xmlNode`"""
        if type(xmlnode_or_cond) is str:
            xmlnode_or_cond=unicode(xmlnode_or_cond,"utf-8")
        self.xmlnode=None
        self.borrowed=0
        if isinstance(xmlnode_or_cond,libxml2.xmlNode):
            self.__from_xml(xmlnode_or_cond,ns,copy,parent)
        elif isinstance(xmlnode_or_cond,ErrorNode):
            if not copy:
                raise TypeError, "ErrorNodes may only be copied"
            self.ns=from_utf8(xmlnode_or_cond.ns.getContent())
            self.xmlnode=xmlnode_or_cond.xmlnode.docCopyNode(common_doc,1)
            if not parent:
                parent=common_root
            parent.addChild(self.xmlnode)
        elif ns is None:
            raise ValueError, "Condition namespace not given"
        else:
            if parent:
                self.xmlnode=parent.newChild(common_ns,"error",None)
                self.borrowed=1
            else:
                self.xmlnode=common_root.newChild(common_ns,"error",None)
            cond=self.xmlnode.newChild(None,to_utf8(xmlnode_or_cond),None)
            ns=cond.newNs(ns,None)
            cond.setNs(ns)
            self.ns=from_utf8(ns.getContent())

    def __from_xml(self,xmlnode,ns,copy,parent):
        """Initialize an ErrorNode object from an XML node.

        :Parameters:
            - `xmlnode`: XML node to be wrapped into this object.
            - `ns`: XML namespace URI of the error condition element (to be
              used when the provided node has no namespace).
            - `copy`: When `True` then the XML node will be copied,
              otherwise it is only borrowed.
            - `parent`: Parent node for the XML node to be copied or created.
        :Types:
            - `xmlnode`: `libxml2.xmlNode`
            - `ns`: `unicode`
            - `copy`: `bool`
            - `parent`: `libxml2.xmlNode`"""
        if not ns:
            ns=None
            c=xmlnode.children
            while c:
                ns=c.ns().getContent()
                if ns in (STREAM_ERROR_NS,STANZA_ERROR_NS):
                    break
                ns=None
                c=c.next
            if ns==None:
                raise ProtocolError, "Bad error namespace"
        self.ns=from_utf8(ns)
        if copy:
            self.xmlnode=xmlnode.docCopyNode(common_doc,1)
            if not parent:
                parent=common_root
            parent.addChild(self.xmlnode)
        else:
            self.xmlnode=xmlnode
            self.borrowed=1
        if copy:
            ns1=xmlnode.ns()
            xmlextra.replace_ns(self.xmlnode, ns1, common_ns)

    def __del__(self):
        if self.xmlnode:
            self.free()

    def free(self):
        """Free the associated XML node."""
        if not self.borrowed:
            self.xmlnode.unlinkNode()
            self.xmlnode.freeNode()
        self.xmlnode=None

    def free_borrowed(self):
        """Free the associated "borrowed" XML node."""
        self.xmlnode=None

    def is_legacy(self):
        """Check if the error node is a legacy error element.

        :return: `True` if it is a legacy error.
        :returntype: `bool`"""
        return not self.xmlnode.hasProp("type")

    def xpath_eval(self,expr,namespaces=None):
        """Evaluate XPath expression on the error element.

        The expression will be evaluated in context where the common namespace
        (the one used for stanza elements, mapped to 'jabber:client',
        'jabber:server', etc.) is bound to prefix "ns" and other namespaces are
        bound accordingly to the `namespaces` list.

        :Parameters:
            - `expr`: the XPath expression.
            - `namespaces`: prefix to namespace mapping.
        :Types:
            - `expr`: `unicode`
            - `namespaces`: `dict`

        :return: the result of the expression evaluation.
        """
        ctxt = common_doc.xpathNewContext()
        ctxt.setContextNode(self.xmlnode)
        ctxt.xpathRegisterNs("ns",to_utf8(self.ns))
        if namespaces:
            for prefix,uri in namespaces.items():
                ctxt.xpathRegisterNs(prefix,uri)
        ret=ctxt.xpathEval(expr)
        ctxt.xpathFreeContext()
        return ret

    def get_condition(self,ns=None):
        """Get the condition element of the error.

        :Parameters:
            - `ns`: namespace URI of the condition element if it is not
              the XMPP namespace of the error element.
        :Types:
            - `ns`: `unicode`

        :return: the condition element or `None`.
        :returntype: `libxml2.xmlNode`"""
        if ns is None:
            ns=self.ns
        c=self.xpath_eval("ns:*")
        if not c:
            self.upgrade()
            c=self.xpath_eval("ns:*")
        if not c:
            return None
        if ns==self.ns and c[0].name=="text":
            if len(c)==1:
                return None
            c=c[1:]
        return c[0]

    def get_text(self):
        """Get the description text from the error element.

        :return: the text provided with the error or `None`.
        :returntype: `unicode`"""
        c=self.xpath_eval("ns:*")
        if not c:
            self.upgrade()
        t=self.xpath_eval("ns:text")
        if not t:
            return None
        return from_utf8(t[0].getContent())

    def add_custom_condition(self,ns,cond,content=None):
        """Add custom condition element to the error.

        :Parameters:
            - `ns`: namespace URI.
            - `cond`: condition name.
            - `content`: content of the element.

        :Types:
            - `ns`: `unicode`
            - `cond`: `unicode`
            - `content`: `unicode`

        :return: the new condition element.
        :returntype: `libxml2.xmlNode`"""
        c=self.xmlnode.newTextChild(None,to_utf8(cond),content)
        ns=c.newNs(to_utf8(ns),None)
        c.setNs(ns)
        return c

    def upgrade(self):
        """Upgrade a legacy error element to the XMPP compliant one.

        Use the error code provided to select the condition and the
        <error/> CDATA for the error text."""

        if not self.xmlnode.hasProp("code"):
            code=None
        else:
            try:
                code=int(self.xmlnode.prop("code"))
            except (ValueError,KeyError):
                code=None

        if code and legacy_codes.has_key(code):
            cond=legacy_codes[code]
        else:
            cond=None

        condition=self.xpath_eval("ns:*")
        if condition:
            return
        elif cond is None:
            condition=self.xmlnode.newChild(None,"undefined-condition",None)
            ns=condition.newNs(to_utf8(self.ns),None)
            condition.setNs(ns)
            condition=self.xmlnode.newChild(None,"unknown-legacy-error",None)
            ns=condition.newNs(PYXMPP_ERROR_NS,None)
            condition.setNs(ns)
        else:
            condition=self.xmlnode.newChild(None,cond,None)
            ns=condition.newNs(to_utf8(self.ns),None)
            condition.setNs(ns)
        txt=self.xmlnode.getContent()
        if txt:
            text=self.xmlnode.newTextChild(None,"text",txt)
            ns=text.newNs(to_utf8(self.ns),None)
            text.setNs(ns)

    def downgrade(self):
        """Downgrade an XMPP error element to the legacy format.

        Add a numeric code attribute according to the condition name."""
        if self.xmlnode.hasProp("code"):
            return
        cond=self.get_condition()
        if not cond:
            return
        cond=cond.name
        if stanza_errors.has_key(cond) and stanza_errors[cond][2]:
            self.xmlnode.setProp("code",to_utf8(stanza_errors[cond][2]))

    def serialize(self):
        """Serialize the element node.

        :return: serialized element in UTF-8 encoding.
        :returntype: `str`"""
        return self.xmlnode.serialize(encoding="utf-8")

class StreamErrorNode(ErrorNode):
    """Stream error element."""
    def __init__(self,xmlnode_or_cond,copy=1,parent=None):
        """Initialize a StreamErrorNode object.

        :Parameters:
            - `xmlnode_or_cond`: XML node to be wrapped into this object
              or the primary (defined by XMPP specification) error condition name.
            - `copy`: When `True` then the XML node will be copied,
              otherwise it is only borrowed.
            - `parent`: Parent node for the XML node to be copied or created.
        :Types:
            - `xmlnode_or_cond`: `libxml2.xmlNode` or `unicode`
            - `copy`: `bool`
            - `parent`: `libxml2.xmlNode`"""
        if type(xmlnode_or_cond) is str:
            xmlnode_or_cond = xmlnode_or_cond.decode("utf-8")
        if type(xmlnode_or_cond) is unicode:
            if not stream_errors.has_key(xmlnode_or_cond):
                raise ValueError, "Bad error condition"
        ErrorNode.__init__(self,xmlnode_or_cond,STREAM_ERROR_NS,copy=copy,parent=parent)

    def get_message(self):
        """Get the message for the error.

        :return: the error message.
        :returntype: `unicode`"""
        cond=self.get_condition()
        if not cond:
            self.upgrade()
            cond=self.get_condition()
            if not cond:
                return None
        cond=cond.name
        if not stream_errors.has_key(cond):
            return None
        return stream_errors[cond][0]

class StanzaErrorNode(ErrorNode):
    """Stanza error element."""
    def __init__(self,xmlnode_or_cond,error_type=None,copy=1,parent=None):
        """Initialize a StreamErrorNode object.

        :Parameters:
            - `xmlnode_or_cond`: XML node to be wrapped into this object
              or the primary (defined by XMPP specification) error condition name.
            - `error_type`: type of the error, one of: 'cancel', 'continue',
              'modify', 'auth', 'wait'.
            - `copy`: When `True` then the XML node will be copied,
              otherwise it is only borrowed.
            - `parent`: Parent node for the XML node to be copied or created.
        :Types:
            - `xmlnode_or_cond`: `libxml2.xmlNode` or `unicode`
            - `error_type`: `unicode`
            - `copy`: `bool`
            - `parent`: `libxml2.xmlNode`"""
        if type(xmlnode_or_cond) is str:
            xmlnode_or_cond=unicode(xmlnode_or_cond,"utf-8")
        if type(xmlnode_or_cond) is unicode:
            if not stanza_errors.has_key(xmlnode_or_cond):
                raise ValueError, "Bad error condition"

        ErrorNode.__init__(self,xmlnode_or_cond,STANZA_ERROR_NS,copy=copy,parent=parent)

        if type(xmlnode_or_cond) is unicode:
            if error_type is None:
                error_type=stanza_errors[xmlnode_or_cond][1]
            self.xmlnode.setProp("type",to_utf8(error_type))

    def get_type(self):
        """Get the error type.

        :return: type of the error.
        :returntype: `unicode`"""
        if not self.xmlnode.hasProp("type"):
            self.upgrade()
        return from_utf8(self.xmlnode.prop("type"))

    def upgrade(self):
        """Upgrade a legacy error element to the XMPP compliant one.

        Use the error code provided to select the condition and the
        <error/> CDATA for the error text."""
        ErrorNode.upgrade(self)
        if self.xmlnode.hasProp("type"):
            return

        cond=self.get_condition().name
        if stanza_errors.has_key(cond):
            typ=stanza_errors[cond][1]
            self.xmlnode.setProp("type",typ)

    def get_message(self):
        """Get the message for the error.

        :return: the error message.
        :returntype: `unicode`"""
        cond=self.get_condition()
        if not cond:
            self.upgrade()
            cond=self.get_condition()
            if not cond:
                return None
        cond=cond.name
        if not stanza_errors.has_key(cond):
            return None
        return stanza_errors[cond][0]

# vi: sts=4 et sw=4