Colin Watson has proposed merging ~cjwatson/launchpad:stormify-logintoken into
launchpad:master.
Commit message:
Convert LoginToken to Storm
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/+git/launchpad/+merge/446394
--
Your team Launchpad code reviewers is requested to review the proposed merge of
~cjwatson/launchpad:stormify-logintoken into launchpad:master.
diff --git a/lib/lp/registry/model/person.py b/lib/lp/registry/model/person.py
index 53e2358..10f2342 100644
--- a/lib/lp/registry/model/person.py
+++ b/lib/lp/registry/model/person.py
@@ -49,6 +49,7 @@ from storm.expr import (
Desc,
Exists,
In,
+ Is,
Join,
LeftJoin,
Min,
@@ -3178,16 +3179,22 @@ class Person(
@property
def unvalidatedemails(self):
"""See `IPerson`."""
- query = """
- requester = %s
- AND (tokentype=%s OR tokentype=%s)
- AND date_consumed IS NULL
- """ % sqlvalues(
- self.id,
- LoginTokenType.VALIDATEEMAIL,
- LoginTokenType.VALIDATETEAMEMAIL,
+ return sorted(
+ {
+ token.email
+ for token in IStore(LoginToken).find(
+ LoginToken,
+ LoginToken.requester == self,
+ LoginToken.tokentype.is_in(
+ (
+ LoginTokenType.VALIDATEEMAIL,
+ LoginTokenType.VALIDATETEAMEMAIL,
+ )
+ ),
+ Is(LoginToken.date_consumed, None),
+ )
+ }
)
- return sorted({token.email for token in LoginToken.select(query)})
@property
def guessedemails(self):
diff --git a/lib/lp/registry/stories/gpg-coc/xx-gpg-coc.rst b/lib/lp/registry/stories/gpg-coc/xx-gpg-coc.rst
index d5b1fc6..d6f3073 100644
--- a/lib/lp/registry/stories/gpg-coc/xx-gpg-coc.rst
+++ b/lib/lp/registry/stories/gpg-coc/xx-gpg-coc.rst
@@ -227,13 +227,14 @@ token to a fixed value:
>>> from datetime import datetime, timezone
>>> import hashlib
>>> from lp.services.verification.model.logintoken import LoginToken
- >>> logintoken = LoginToken.selectOneBy(
- ... _token=hashlib.sha256(token_value).hexdigest()
+ >>> logintoken = (
+ ... IStore(LoginToken)
+ ... .find(LoginToken, _token=hashlib.sha256(token_value).hexdigest())
+ ... .one()
... )
>>> logintoken.date_created = datetime(
... 2005, 4, 1, 12, 0, 0, tzinfo=timezone.utc
... )
- >>> logintoken.sync()
Back to Sample User. They visit the token URL and is asked to sign some
text to prove they own the key.
@@ -317,8 +318,10 @@ If they sign the text correctly, they are redirected to their home page.
Now that the key has been validated, the login token is consumed:
- >>> consumed_token = LoginToken.selectOneBy(
- ... _token=hashlib.sha256(token_value).hexdigest()
+ >>> consumed_token = (
+ ... IStore(LoginToken)
+ ... .find(LoginToken, _token=hashlib.sha256(token_value).hexdigest())
+ ... .one()
... )
>>> consumed_token.date_consumed is not None
True
diff --git a/lib/lp/services/verification/model/logintoken.py b/lib/lp/services/verification/model/logintoken.py
index 5aafa18..572cce2 100644
--- a/lib/lp/services/verification/model/logintoken.py
+++ b/lib/lp/services/verification/model/logintoken.py
@@ -10,7 +10,9 @@ import hashlib
from datetime import timezone
import six
-from storm.expr import And
+from storm.expr import And, Is
+from storm.properties import DateTime, Int, Unicode
+from storm.references import Reference
from zope.component import getUtility
from zope.interface import implementer
@@ -20,15 +22,9 @@ from lp.registry.interfaces.gpg import IGPGKeySet
from lp.registry.interfaces.person import IPersonSet
from lp.services.config import config
from lp.services.database.constants import UTC_NOW
-from lp.services.database.datetimecol import UtcDateTimeCol
from lp.services.database.enumcol import DBEnum
from lp.services.database.interfaces import IPrimaryStore, IStore
-from lp.services.database.sqlbase import SQLBase, sqlvalues
-from lp.services.database.sqlobject import (
- ForeignKey,
- SQLObjectNotFound,
- StringCol,
-)
+from lp.services.database.stormbase import StormBase
from lp.services.gpg.interfaces import IGPGHandler
from lp.services.mail.helpers import get_email_template
from lp.services.mail.sendmail import format_address, simple_sendmail
@@ -44,35 +40,51 @@ MAIL_APP = "services/verification"
@implementer(ILoginToken)
-class LoginToken(SQLBase):
- _table = "LoginToken"
-
- redirection_url = StringCol(default=None)
- requester = ForeignKey(dbName="requester", foreignKey="Person")
- requesteremail = StringCol(
- dbName="requesteremail", notNull=False, default=None
+class LoginToken(StormBase):
+ __storm_table__ = "LoginToken"
+
+ id = Int(primary=True)
+ redirection_url = Unicode(default=None)
+ requester_id = Int(name="requester")
+ requester = Reference(requester_id, "Person.id")
+ requesteremail = Unicode(
+ name="requesteremail", allow_none=True, default=None
)
- email = StringCol(dbName="email", notNull=True)
+ email = Unicode(name="email", allow_none=False)
# The hex SHA-256 hash of the token.
- _token = StringCol(dbName="token", unique=True)
+ _token = Unicode(name="token")
tokentype = DBEnum(name="tokentype", allow_none=False, enum=LoginTokenType)
- date_created = UtcDateTimeCol(dbName="created", notNull=True)
- fingerprint = StringCol(dbName="fingerprint", notNull=False, default=None)
- date_consumed = UtcDateTimeCol(default=None)
+ date_created = DateTime(
+ name="created", allow_none=False, tzinfo=timezone.utc
+ )
+ fingerprint = Unicode(name="fingerprint", allow_none=True, default=None)
+ date_consumed = DateTime(default=None, tzinfo=timezone.utc)
password = "" # Quick fix for Bug #2481
title = "Launchpad Email Verification"
- def __init__(self, *args, **kwargs):
- token = kwargs.pop("token", None)
+ def __init__(
+ self,
+ email,
+ tokentype,
+ redirection_url=None,
+ requester=None,
+ requesteremail=None,
+ token=None,
+ fingerprint=None,
+ ):
+ super().__init__()
+ self.email = email
+ self.tokentype = tokentype
+ self.redirection_url = redirection_url
+ self.requester = requester
+ self.requesteremail = requesteremail
if token is not None:
self._plaintext_token = token
- kwargs["_token"] = hashlib.sha256(
- token.encode("UTF-8")
- ).hexdigest()
- super().__init__(*args, **kwargs)
+ self._token = hashlib.sha256(token.encode("UTF-8")).hexdigest()
+ self.fingerprint = fingerprint
_plaintext_token = None
@@ -267,6 +279,10 @@ class LoginToken(SQLBase):
self.consume()
return lpkey, new
+ def destroySelf(self):
+ """See `ILoginToken`."""
+ IStore(self).remove(self)
+
@implementer(ILoginTokenSet)
class LoginTokenSet:
@@ -275,10 +291,10 @@ class LoginTokenSet:
def get(self, id, default=None):
"""See ILoginTokenSet."""
- try:
- return LoginToken.get(id)
- except SQLObjectNotFound:
+ token = IStore(LoginToken).get(LoginToken, id)
+ if token is None:
return default
+ return token
def searchByEmailRequesterAndType(
self, email, requester, type, consumed=None
@@ -337,18 +353,20 @@ class LoginTokenSet:
def getPendingGPGKeys(self, requesterid=None):
"""See ILoginTokenSet."""
- query = (
- "date_consumed IS NULL AND "
- "(tokentype = %s OR tokentype = %s) "
- % sqlvalues(
- LoginTokenType.VALIDATEGPG, LoginTokenType.VALIDATESIGNONLYGPG
- )
- )
+ clauses = [
+ Is(LoginToken.date_consumed, None),
+ LoginToken.tokentype.is_in(
+ (
+ LoginTokenType.VALIDATEGPG,
+ LoginTokenType.VALIDATESIGNONLYGPG,
+ )
+ ),
+ ]
if requesterid:
- query += "AND requester=%s" % requesterid
+ clauses.append(LoginToken.requester == requesterid)
- return LoginToken.select(query)
+ return IStore(LoginToken).find(LoginToken, *clauses)
def deleteByFingerprintRequesterAndType(
self, fingerprint, requester, type
@@ -383,7 +401,6 @@ class LoginTokenSet:
email=email,
token=token,
tokentype=tokentype,
- created=UTC_NOW,
fingerprint=fingerprint,
redirection_url=redirection_url,
)
_______________________________________________
Mailing list: https://launchpad.net/~launchpad-reviewers
Post to : launchpad-reviewers@lists.launchpad.net
Unsubscribe : https://launchpad.net/~launchpad-reviewers
More help : https://help.launchpad.net/ListHelp
