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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Database class for table ArchiveAuthToken."""
__metaclass__ = type
__all__ = [
'ArchiveAuthToken',
]
from lazr.uri import URI
import pytz
from storm.locals import (
DateTime,
Int,
Reference,
Storm,
Unicode,
)
from storm.store import Store
from zope.component import getUtility
from zope.interface import implements
from canonical.database.constants import UTC_NOW
from canonical.launchpad.webapp.interfaces import (
DEFAULT_FLAVOR,
IStoreSelector,
MAIN_STORE,
)
from lp.soyuz.interfaces.archiveauthtoken import (
IArchiveAuthToken,
IArchiveAuthTokenSet,
)
class ArchiveAuthToken(Storm):
"""See `IArchiveAuthToken`."""
implements(IArchiveAuthToken)
__storm_table__ = 'ArchiveAuthToken'
id = Int(primary=True)
archive_id = Int(name='archive', allow_none=False)
archive = Reference(archive_id, 'Archive.id')
person_id = Int(name='person', allow_none=False)
person = Reference(person_id, 'Person.id')
date_created = DateTime(
name='date_created', allow_none=False, tzinfo=pytz.UTC)
date_deactivated = DateTime(
name='date_deactivated', allow_none=True, tzinfo=pytz.UTC)
token = Unicode(name='token', allow_none=False)
def deactivate(self):
"""See `IArchiveAuthTokenSet`."""
self.date_deactivated = UTC_NOW
@property
def archive_url(self):
"""Return a custom archive url for basic authentication."""
normal_url = URI(self.archive.archive_url)
auth_url = normal_url.replace(
userinfo="%s:%s" %(self.person.name, self.token))
return str(auth_url)
class ArchiveAuthTokenSet:
"""See `IArchiveAuthTokenSet`."""
implements(IArchiveAuthTokenSet)
title = "Archive Tokens in Launchpad"
def get(self, token_id):
"""See `IArchiveAuthTokenSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
return store.get(ArchiveAuthToken, token_id)
def getByToken(self, token):
"""See `IArchiveAuthTokenSet`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
return store.find(
ArchiveAuthToken,
ArchiveAuthToken.token == token).one()
def getByArchive(self, archive):
"""See `IArchiveAuthTokenSet`."""
store = Store.of(archive)
return store.find(
ArchiveAuthToken,
ArchiveAuthToken.archive == archive,
ArchiveAuthToken.date_deactivated == None)
def getActiveTokenForArchiveAndPerson(self, archive, person):
"""See `IArchiveAuthTokenSet`."""
store = Store.of(archive)
return store.find(
ArchiveAuthToken,
ArchiveAuthToken.archive == archive,
ArchiveAuthToken.person == person,
ArchiveAuthToken.date_deactivated == None).one()
|