← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/plugins/launchpad/lp_registration.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-11-04 18:51:39 UTC
  • mfrom: (2961.1.1 trunk)
  • Revision ID: pqm@pqm.ubuntu.com-20071104185139-kaio3sneodg2kp71
Authentication ring implementation (read-only)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_registration.py
32
32
export BZR_LP_XMLRPC_URL=http://xmlrpc.staging.launchpad.net/bazaar/
33
33
'''
34
34
 
35
 
class InvalidLaunchpadInstance(errors.BzrError):
36
 
 
37
 
    _fmt = "%(lp_instance)s is not a valid Launchpad instance."
38
 
 
39
 
    def __init__(self, lp_instance):
40
 
        errors.BzrError.__init__(self, lp_instance=lp_instance)
41
 
 
42
 
 
43
35
class LaunchpadService(object):
44
36
    """A service to talk to Launchpad via XMLRPC.
45
 
 
 
37
    
46
38
    See http://bazaar-vcs.org/Specs/LaunchpadRpc for the methods we can call.
47
39
    """
48
40
 
49
 
    # NB: these should always end in a slash to avoid xmlrpclib appending
 
41
    # NB: this should always end in a slash to avoid xmlrpclib appending
50
42
    # '/RPC2'
51
 
    # We use edge as the default because:
52
 
    # Beta users get redirected to it
53
 
    # All users can use it
54
 
    # There is a bug in the launchpad side where redirection causes an OOPS.
55
 
    LAUNCHPAD_INSTANCE = {
56
 
        'production': 'https://xmlrpc.launchpad.net/bazaar/',
57
 
        'edge': 'https://xmlrpc.edge.launchpad.net/bazaar/',
58
 
        'staging': 'https://xmlrpc.staging.launchpad.net/bazaar/',
59
 
        'demo': 'https://xmlrpc.demo.launchpad.net/bazaar/',
60
 
        'dev': 'http://xmlrpc.launchpad.dev/bazaar/',
61
 
        }
62
 
    DEFAULT_SERVICE_URL = LAUNCHPAD_INSTANCE['edge']
 
43
    DEFAULT_SERVICE_URL = 'https://xmlrpc.launchpad.net/bazaar/'
63
44
 
64
45
    transport = None
65
46
    registrant_email = None
66
47
    registrant_password = None
67
48
 
68
49
 
69
 
    def __init__(self, transport=None, lp_instance=None):
 
50
    def __init__(self, transport=None):
70
51
        """Construct a new service talking to the launchpad rpc server"""
71
 
        self._lp_instance = lp_instance
72
52
        if transport is None:
73
53
            uri_type = urllib.splittype(self.service_url)[0]
74
54
            if uri_type == 'https':
89
69
        key = 'BZR_LP_XMLRPC_URL'
90
70
        if key in os.environ:
91
71
            return os.environ[key]
92
 
        elif self._lp_instance is not None:
93
 
            try:
94
 
                return self.LAUNCHPAD_INSTANCE[self._lp_instance]
95
 
            except KeyError:
96
 
                raise InvalidLaunchpadInstance(self._lp_instance)
97
72
        else:
98
73
            return self.DEFAULT_SERVICE_URL
99
74
 
100
 
    def get_proxy(self, authenticated):
 
75
    def get_proxy(self):
101
76
        """Return the proxy for XMLRPC requests."""
102
 
        if authenticated:
103
 
            # auth info must be in url
104
 
            # TODO: if there's no registrant email perhaps we should
105
 
            # just connect anonymously?
106
 
            scheme, hostinfo, path = urlsplit(self.service_url)[:3]
107
 
            if '@' in hostinfo:
108
 
                raise AssertionError(hostinfo)
109
 
            if self.registrant_email is None:
110
 
                raise AssertionError()
111
 
            if self.registrant_password is None:
112
 
                raise AssertionError()
113
 
            # TODO: perhaps fully quote the password to make it very slightly
114
 
            # obscured
115
 
            # TODO: can we perhaps add extra Authorization headers
116
 
            # directly to the request, rather than putting this into
117
 
            # the url?  perhaps a bit more secure against accidentally
118
 
            # revealing it.  std66 s3.2.1 discourages putting the
119
 
            # password in the url.
120
 
            hostinfo = '%s:%s@%s' % (urllib.quote(self.registrant_email),
121
 
                                     urllib.quote(self.registrant_password),
122
 
                                     hostinfo)
123
 
            url = urlunsplit((scheme, hostinfo, path, '', ''))
124
 
        else:
125
 
            url = self.service_url
 
77
        # auth info must be in url
 
78
        # TODO: if there's no registrant email perhaps we should just connect
 
79
        # anonymously?
 
80
        scheme, hostinfo, path = urlsplit(self.service_url)[:3]
 
81
        assert '@' not in hostinfo
 
82
        assert self.registrant_email is not None
 
83
        assert self.registrant_password is not None
 
84
        # TODO: perhaps fully quote the password to make it very slightly
 
85
        # obscured
 
86
        # TODO: can we perhaps add extra Authorization headers directly to the 
 
87
        # request, rather than putting this into the url?  perhaps a bit more 
 
88
        # secure against accidentally revealing it.  std66 s3.2.1 discourages putting
 
89
        # the password in the url.
 
90
        hostinfo = '%s:%s@%s' % (urllib.quote(self.registrant_email),
 
91
                                 urllib.quote(self.registrant_password),
 
92
                                 hostinfo)
 
93
        url = urlunsplit((scheme, hostinfo, path, '', ''))
126
94
        return xmlrpclib.ServerProxy(url, transport=self.transport)
127
95
 
128
96
    def gather_user_credentials(self):
129
97
        """Get the password from the user."""
130
 
        the_config = config.GlobalConfig()
131
 
        self.registrant_email = the_config.user_email()
 
98
        config = config.GlobalConfig()
 
99
        self.registrant_email = config.user_email()
132
100
        if self.registrant_password is None:
133
101
            auth = config.AuthenticationConfig()
134
102
            scheme, hostinfo = urlsplit(self.service_url)[:2]
137
105
            # We will reuse http[s] credentials if we can, prompt user
138
106
            # otherwise
139
107
            self.registrant_password = auth.get_password(scheme, hostinfo,
140
 
                                                         self.registrant_email,
141
108
                                                         prompt=prompt)
142
109
 
143
 
    def send_request(self, method_name, method_params, authenticated):
144
 
        proxy = self.get_proxy(authenticated)
 
110
    def send_request(self, method_name, method_params):
 
111
        proxy = self.get_proxy()
 
112
        assert method_name
145
113
        method = getattr(proxy, method_name)
146
114
        try:
147
115
            result = method(*method_params)
166
134
 
167
135
    # Set this to the XMLRPC method name.
168
136
    _methodname = None
169
 
    _authenticated = True
170
137
 
171
138
    def _request_params(self):
172
139
        """Return the arguments to pass to the method"""
178
145
        :param service: LaunchpadService indicating where to send
179
146
            the request and the authentication credentials.
180
147
        """
181
 
        return service.send_request(self._methodname, self._request_params(),
182
 
                                    self._authenticated)
 
148
        return service.send_request(self._methodname, self._request_params())
183
149
 
184
150
 
185
151
class DryRunLaunchpadService(LaunchpadService):
188
154
    The dummy service does not need authentication.
189
155
    """
190
156
 
191
 
    def send_request(self, method_name, method_params, authenticated):
 
157
    def send_request(self, method_name, method_params):
192
158
        pass
193
159
 
194
160
    def gather_user_credentials(self):
207
173
                 author_email='',
208
174
                 product_name='',
209
175
                 ):
210
 
        if not branch_url:
211
 
            raise errors.InvalidURL(branch_url, "You need to specify a non-empty branch URL.")
 
176
        assert branch_url
212
177
        self.branch_url = branch_url
213
178
        if branch_name:
214
179
            self.branch_name = branch_name
242
207
    _methodname = 'link_branch_to_bug'
243
208
 
244
209
    def __init__(self, branch_url, bug_id):
 
210
        assert branch_url
245
211
        self.bug_id = bug_id
246
212
        self.branch_url = branch_url
247
213
 
250
216
        # This must match the parameter tuple expected by Launchpad for this
251
217
        # method
252
218
        return (self.branch_url, self.bug_id, '')
253
 
 
254
 
 
255
 
class ResolveLaunchpadPathRequest(BaseRequest):
256
 
    """Request to resolve the path component of an lp: URL."""
257
 
 
258
 
    _methodname = 'resolve_lp_path'
259
 
    _authenticated = False
260
 
 
261
 
    def __init__(self, path):
262
 
        if not path:
263
 
            raise errors.InvalidURL(path=path,
264
 
                                    extra="You must specify a product.")
265
 
        self.path = path
266
 
 
267
 
    def _request_params(self):
268
 
        """Return xmlrpc request parameters"""
269
 
        return (self.path,)