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
|
# Copyright 2008 Canonical Ltd. All rights reserved.
"""Implementations of the XML-RPC APIs for codehosting."""
__metaclass__ = type
__all__ = [
'BranchFileSystem',
'BranchPuller',
'datetime_from_tuple',
'iter_split',
]
import datetime
import pytz
from bzrlib.urlutils import escape, unescape
from zope.component import getUtility
from zope.interface import implements
from zope.security.interfaces import Unauthorized
from zope.security.proxy import removeSecurityProxy
from canonical.launchpad.ftests import login_person, logout
from lp.code.interfaces.branch import (
BranchType, BranchCreationException, UnknownBranchTypeError)
from lp.code.interfaces.branchlookup import IBranchLookup
from lp.code.interfaces.branchnamespace import (
InvalidNamespace, lookup_branch_namespace, split_unique_name)
from lp.code.interfaces import branchpuller
from lp.code.interfaces.codehosting import (
BRANCH_TRANSPORT, CONTROL_TRANSPORT, IBranchFileSystem, IBranchPuller,
LAUNCHPAD_ANONYMOUS, LAUNCHPAD_SERVICES)
from lp.registry.interfaces.person import IPersonSet, NoSuchPerson
from lp.registry.interfaces.product import NoSuchProduct
from lp.services.scripts.interfaces.scriptactivity import IScriptActivitySet
from canonical.launchpad.validators import LaunchpadValidationError
from canonical.launchpad.webapp import LaunchpadXMLRPCView
from canonical.launchpad.webapp.authorization import check_permission
from canonical.launchpad.webapp.interfaces import (
NameLookupFailed, NotFoundError)
from canonical.launchpad.xmlrpc import faults
from canonical.launchpad.xmlrpc.helpers import return_fault
from canonical.launchpad.webapp.interaction import Participation
UTC = pytz.timezone('UTC')
class BranchPuller(LaunchpadXMLRPCView):
"""See `IBranchPuller`."""
implements(IBranchPuller)
def _getBranchPullInfo(self, branch):
"""Return information the branch puller needs to pull this branch.
This is outside of the IBranch interface so that the authserver can
access the information without logging in as a particular user.
:return: (id, url, unique_name, default_stacked_on_url), where 'id'
is the branch database ID, 'url' is the URL to pull from,
'unique_name' is the `unique_name` property and
'default_stacked_on_url' is the URL of the branch to stack on by
default (normally of the form '/~foo/bar/baz'). If there is no
default stacked-on branch, then it's ''.
"""
branch = removeSecurityProxy(branch)
if branch.branch_type == BranchType.REMOTE:
raise AssertionError(
'Remote branches should never be in the pull queue.')
default_branch = branch.target.default_stacked_on_branch
if default_branch is None:
default_branch = ''
elif (branch.branch_type == BranchType.MIRRORED
and default_branch.private):
default_branch = ''
else:
default_branch = '/' + default_branch.unique_name
return (
branch.id, branch.getPullURL(), branch.unique_name,
default_branch)
def getBranchPullQueue(self, branch_type):
"""See `IBranchPuller`."""
try:
branch_type = BranchType.items[branch_type]
except KeyError:
raise UnknownBranchTypeError(
'Unknown branch type: %r' % (branch_type,))
branches = getUtility(branchpuller.IBranchPuller).getPullQueue(
branch_type)
return [self._getBranchPullInfo(branch) for branch in branches]
def acquireBranchToPull(self):
"""See `IBranchPuller`."""
branch = getUtility(branchpuller.IBranchPuller).acquireBranchToPull()
if branch is not None:
default_branch = branch.target.default_stacked_on_branch
if default_branch:
default_branch_name = default_branch.unique_name
else:
default_branch_name = ''
return (branch.id, branch.getPullURL(), branch.unique_name,
default_branch_name, branch.branch_type.name)
else:
return ()
def mirrorComplete(self, branch_id, last_revision_id):
"""See `IBranchPuller`."""
branch = getUtility(IBranchLookup).get(branch_id)
if branch is None:
return faults.NoBranchWithID(branch_id)
# See comment in startMirroring.
branch = removeSecurityProxy(branch)
branch.mirrorComplete(last_revision_id)
branches = branch.getStackedBranchesWithIncompleteMirrors()
for stacked_branch in branches:
stacked_branch.requestMirror()
return True
def mirrorFailed(self, branch_id, reason):
"""See `IBranchPuller`."""
branch = getUtility(IBranchLookup).get(branch_id)
if branch is None:
return faults.NoBranchWithID(branch_id)
# See comment in startMirroring.
removeSecurityProxy(branch).mirrorFailed(reason)
return True
def recordSuccess(self, name, hostname, started_tuple, completed_tuple):
"""See `IBranchPuller`."""
date_started = datetime_from_tuple(started_tuple)
date_completed = datetime_from_tuple(completed_tuple)
getUtility(IScriptActivitySet).recordSuccess(
name=name, date_started=date_started,
date_completed=date_completed, hostname=hostname)
return True
def startMirroring(self, branch_id):
"""See `IBranchPuller`."""
branch = getUtility(IBranchLookup).get(branch_id)
if branch is None:
return faults.NoBranchWithID(branch_id)
# The puller runs as no user and may pull private branches. We need to
# bypass Zope's security proxy to set the mirroring information.
removeSecurityProxy(branch).startMirroring()
return True
def setStackedOn(self, branch_id, stacked_on_location):
"""See `IBranchPuller`."""
# We don't want the security proxy on the branch set because this
# method should be able to see all branches and set stacking
# information on any of them.
branch_set = removeSecurityProxy(getUtility(IBranchLookup))
if stacked_on_location == '':
stacked_on_branch = None
else:
if stacked_on_location.startswith('/'):
stacked_on_branch = branch_set.getByUniqueName(
stacked_on_location.strip('/'))
else:
stacked_on_branch = branch_set.getByUrl(
stacked_on_location.rstrip('/'))
if stacked_on_branch is None:
return faults.NoSuchBranch(stacked_on_location)
stacked_branch = branch_set.get(branch_id)
if stacked_branch is None:
return faults.NoBranchWithID(branch_id)
stacked_branch.stacked_on = stacked_on_branch
return True
def datetime_from_tuple(time_tuple):
"""Create a datetime from a sequence that quacks like time.struct_time.
The tm_isdst is (index 8) is ignored. The created datetime uses
tzinfo=UTC.
"""
[year, month, day, hour, minute, second, unused, unused, unused] = (
time_tuple)
return datetime.datetime(
year, month, day, hour, minute, second, tzinfo=UTC)
def run_with_login(login_id, function, *args, **kwargs):
"""Run 'function' logged in with 'login_id'.
The first argument passed to 'function' will be the Launchpad
`Person` object corresponding to 'login_id'.
The exception is when the requesting login ID is `LAUNCHPAD_SERVICES`. In
that case, we'll pass through the `LAUNCHPAD_SERVICES` variable and the
method will do whatever security proxy hackery is required to provide read
privileges to the Launchpad services.
"""
if login_id == LAUNCHPAD_SERVICES or login_id == LAUNCHPAD_ANONYMOUS:
# Don't pass in an actual user. Instead pass in LAUNCHPAD_SERVICES
# and expect `function` to use `removeSecurityProxy` or similar.
return function(login_id, *args, **kwargs)
if isinstance(login_id, basestring):
requester = getUtility(IPersonSet).getByName(login_id)
else:
requester = getUtility(IPersonSet).get(login_id)
if requester is None:
raise NotFoundError("No person with id %s." % login_id)
# XXX gary 21-Oct-2008 bug 285808
# We should reconsider using a ftest helper for production code. For now,
# we explicitly keep the code from using a test request by using a basic
# participation.
login_person(requester, Participation())
try:
return function(requester, *args, **kwargs)
finally:
logout()
class BranchFileSystem(LaunchpadXMLRPCView):
"""See `IBranchFileSystem`."""
implements(IBranchFileSystem)
def createBranch(self, login_id, branch_path):
"""See `IBranchFileSystem`."""
def create_branch(requester):
if not branch_path.startswith('/'):
return faults.InvalidPath(branch_path)
escaped_path = unescape(branch_path.strip('/')).encode('utf-8')
try:
namespace_name, branch_name = split_unique_name(escaped_path)
except ValueError:
return faults.PermissionDenied(
"Cannot create branch at '%s'" % branch_path)
try:
namespace = lookup_branch_namespace(namespace_name)
except InvalidNamespace:
return faults.PermissionDenied(
"Cannot create branch at '%s'" % branch_path)
except NoSuchPerson, e:
return faults.NotFound(
"User/team '%s' does not exist." % e.name)
except NoSuchProduct, e:
return faults.NotFound(
"Project '%s' does not exist." % e.name)
except NameLookupFailed, e:
return faults.NotFound(str(e))
try:
branch = namespace.createBranch(
BranchType.HOSTED, branch_name, requester)
except (BranchCreationException, LaunchpadValidationError), e:
return faults.PermissionDenied(str(e))
else:
return branch.id
return run_with_login(login_id, create_branch)
def _canWriteToBranch(self, requester, branch):
"""Can `requester` write to `branch`?"""
if requester == LAUNCHPAD_SERVICES:
return False
return (branch.branch_type == BranchType.HOSTED
and check_permission('launchpad.Edit', branch))
def requestMirror(self, login_id, branchID):
"""See `IBranchFileSystem`."""
def request_mirror(requester):
branch = getUtility(IBranchLookup).get(branchID)
# We don't really care who requests a mirror of a branch.
branch.requestMirror()
return True
return run_with_login(login_id, request_mirror)
def _serializeBranch(self, requester, branch, trailing_path):
if requester == LAUNCHPAD_SERVICES:
branch = removeSecurityProxy(branch)
try:
branch_id = branch.id
except Unauthorized:
raise faults.PermissionDenied()
if branch.branch_type == BranchType.REMOTE:
return None
return (
BRANCH_TRANSPORT,
{'id': branch_id,
'writable': self._canWriteToBranch(requester, branch)},
trailing_path)
def _serializeControlDirectory(self, requester, product_path,
trailing_path):
try:
namespace = lookup_branch_namespace(
unescape(product_path).encode('utf-8'))
except (InvalidNamespace, NotFoundError):
return
if not ('.bzr' == trailing_path or trailing_path.startswith('.bzr/')):
# '.bzr' is OK, '.bzr/foo' is OK, '.bzrfoo' is not.
return
default_branch = namespace.target.default_stacked_on_branch
if default_branch is None:
return
try:
unique_name = default_branch.unique_name
except Unauthorized:
return
return (
CONTROL_TRANSPORT,
{'default_stack_on': escape('/' + unique_name)},
trailing_path)
def translatePath(self, requester_id, path):
"""See `IBranchFileSystem`."""
@return_fault
def translate_path(requester):
if not path.startswith('/'):
return faults.InvalidPath(path)
stripped_path = path.strip('/')
for first, second in iter_split(stripped_path, '/'):
# Is it a branch?
branch = getUtility(IBranchLookup).getByUniqueName(
unescape(first).encode('utf-8'))
if branch is not None:
branch = self._serializeBranch(requester, branch, second)
if branch is None:
break
return branch
# Is it a product control directory?
product = self._serializeControlDirectory(
requester, first, second)
if product is not None:
return product
raise faults.PathTranslationError(path)
return run_with_login(requester_id, translate_path)
def iter_split(string, splitter):
"""Iterate over ways to split 'string' in two with 'splitter'.
If 'string' is empty, then yield nothing. Otherwise, yield tuples like
('a/b/c', ''), ('a/b', 'c'), ('a', 'b/c') for a string 'a/b/c' and a
splitter '/'.
The tuples are yielded such that the first tuple has everything in the
first tuple. With each iteration, the first element gets smaller and the
second gets larger. It stops iterating just before it would have to yield
('', 'a/b/c').
"""
if string == '':
return
tokens = string.split(splitter)
for i in reversed(range(1, len(tokens) + 1)):
yield splitter.join(tokens[:i]), splitter.join(tokens[i:])
|