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
|
# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com>
#
# This file is part of nbxmpp.
#
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations
from typing import TYPE_CHECKING
from nbxmpp.errors import ChangePasswordStanzaError
from nbxmpp.errors import RegisterStanzaError
from nbxmpp.errors import StanzaError
from nbxmpp.modules.base import BaseModule
from nbxmpp.modules.register.util import _make_password_change_request
from nbxmpp.modules.register.util import _make_password_change_with_form
from nbxmpp.modules.register.util import _make_register_form
from nbxmpp.modules.register.util import _make_unregister_request
from nbxmpp.modules.register.util import _parse_register_data
from nbxmpp.modules.util import process_response
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import Iq
from nbxmpp.protocol import JID
from nbxmpp.task import iq_request_task
from nbxmpp.util import get_form
if TYPE_CHECKING:
from nbxmpp.client import Client
class Register(BaseModule):
def __init__(self, client: Client) -> None:
BaseModule.__init__(self, client)
self._client = client
self.handlers = []
@iq_request_task
def unregister(self, jid: JID | None = None):
_task = yield
response = yield _make_unregister_request(jid)
yield process_response(response)
@iq_request_task
def request_register_form(self, jid: JID | None = None):
_task = yield
if jid is None:
jid = self._client.domain
response = yield Iq("get", Namespace.REGISTER, to=jid)
if response.isError():
raise StanzaError(response)
yield _parse_register_data(response)
@iq_request_task
def submit_register_form(self, form, jid: JID | None = None):
_task = yield
if jid is None:
jid = self._client.domain
response = yield _make_register_form(jid, form)
if not response.isError():
yield process_response(response)
else:
data = _parse_register_data(response)
raise RegisterStanzaError(response, data)
@iq_request_task
def change_password(self, password: str):
_task = yield
response = yield _make_password_change_request(
self._client.domain, self._client.username, password
)
if not response.isError():
yield process_response(response)
else:
query = response.getQuery()
if query is None:
raise StanzaError(response)
form = get_form(query, "jabber:iq:register:changepassword")
if form is None or response.getType() != "modify":
raise StanzaError(response)
raise ChangePasswordStanzaError(response, form)
@iq_request_task
def change_password_with_form(self, form):
_task = yield
response = yield _make_password_change_with_form(self._client.domain, form)
yield process_response(response)
|
