139 lines
6.0 KiB
Python
Executable File
139 lines
6.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""2925 account-pool helper. Never prints passwords."""
|
|
import argparse, json, os, random, re, stat, string, time
|
|
from pathlib import Path
|
|
|
|
ACCOUNTS_FILE = Path(os.environ.get('MAIL2925_ACCOUNTS_FILE', str(Path.home()/'.hermes/secrets/mail2925_accounts.txt'))).expanduser()
|
|
STATE_FILE = Path(os.environ.get('MAIL2925_STATE_FILE', str(Path.home()/'.hermes/state/mail2925_accounts_state.json'))).expanduser()
|
|
COOLDOWN_SECONDS = int(os.environ.get('MAIL2925_LIMIT_COOLDOWN_SECONDS', str(24*60*60)))
|
|
SUFFIX_LEN = int(os.environ.get('MAIL2925_ALIAS_SUFFIX_LEN', '6'))
|
|
|
|
EMAIL_RE = re.compile(r'^[^@\s]+@2925\.com$', re.I)
|
|
|
|
|
|
def load_accounts(path=ACCOUNTS_FILE):
|
|
text = path.read_text()
|
|
out=[]; seen=set()
|
|
for raw in text.splitlines():
|
|
line=raw.strip()
|
|
if not line or line.lower() == '邮箱----密码':
|
|
continue
|
|
if '----' not in line:
|
|
raise ValueError(f'bad account line: {line[:20]}...')
|
|
email,pwd=[x.strip() for x in line.split('----',1)]
|
|
email=email.lower()
|
|
if not EMAIL_RE.match(email):
|
|
raise ValueError(f'bad 2925 email: {email}')
|
|
if not pwd:
|
|
raise ValueError(f'missing password: {email}')
|
|
if email in seen:
|
|
continue
|
|
seen.add(email); out.append({'id': email, 'email': email, 'password': pwd})
|
|
return out
|
|
|
|
|
|
def load_state():
|
|
if not STATE_FILE.exists(): return {'accounts': {}}
|
|
try: return json.loads(STATE_FILE.read_text())
|
|
except Exception: return {'accounts': {}}
|
|
|
|
|
|
def save_state(state):
|
|
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True))
|
|
os.chmod(STATE_FILE, 0o600)
|
|
|
|
|
|
def public_account(a, state):
|
|
s=state.get('accounts',{}).get(a['email'],{})
|
|
now=time.time(); disabled=float(s.get('disabledUntil') or 0)
|
|
status='cooldown' if disabled>now else ('disabled' if s.get('enabled') is False else 'ready')
|
|
return {'email': a['email'], 'status': status, 'lastUsedAt': int(s.get('lastUsedAt') or 0), 'lastLoginAt': int(s.get('lastLoginAt') or 0), 'disabledUntil': int(disabled), 'lastError': s.get('lastError','')}
|
|
|
|
|
|
def available_accounts(accounts, state):
|
|
now=time.time(); out=[]
|
|
for a in accounts:
|
|
s=state.get('accounts',{}).get(a['email'],{})
|
|
if s.get('enabled') is False: continue
|
|
if float(s.get('disabledUntil') or 0)>now: continue
|
|
out.append(a)
|
|
return out
|
|
|
|
|
|
def pick_account(preferred=None, exclude=None):
|
|
accounts=load_accounts(); state=load_state(); exclude=set(exclude or [])
|
|
pool=[a for a in available_accounts(accounts,state) if a['email'] not in exclude]
|
|
if preferred:
|
|
preferred=preferred.lower().strip()
|
|
for a in pool:
|
|
if a['email']==preferred: return a,state
|
|
if not pool: raise RuntimeError('no available 2925 account')
|
|
def key(a):
|
|
s=state.get('accounts',{}).get(a['email'],{})
|
|
return (float(s.get('lastUsedAt') or 0), a['email'])
|
|
return sorted(pool,key=key)[0], state
|
|
|
|
|
|
def generate_alias_for(account):
|
|
local=account['email'].split('@',1)[0].lower()
|
|
suffix=''.join(random.choice(string.ascii_lowercase+string.digits) for _ in range(SUFFIX_LEN))
|
|
alias=f'{local}{suffix}@2925.com'
|
|
assert alias.startswith(local) and alias.endswith('@2925.com')
|
|
return alias
|
|
|
|
|
|
def mark(email, **patch):
|
|
state=load_state(); d=state.setdefault('accounts',{}).setdefault(email.lower(),{})
|
|
d.update(patch); save_state(state)
|
|
|
|
|
|
def cmd_list(args):
|
|
accounts=load_accounts(); state=load_state()
|
|
print(json.dumps({'count':len(accounts),'accounts':[public_account(a,state) for a in accounts]}, ensure_ascii=False))
|
|
|
|
|
|
def cmd_pick(args):
|
|
a,state=pick_account(args.preferred, args.exclude or [])
|
|
alias=generate_alias_for(a)
|
|
if not args.dry_run:
|
|
mark(a['email'], lastUsedAt=int(time.time()), lastError='')
|
|
print(json.dumps({'email':a['email'],'alias':alias,'baseLocal':a['email'].split('@')[0],'aliasMatchesAccount':alias.startswith(a['email'].split('@')[0]),'dryRun':bool(args.dry_run)}, ensure_ascii=False))
|
|
|
|
|
|
def cmd_limit(args):
|
|
email=args.email.lower().strip(); until=int(time.time()+COOLDOWN_SECONDS)
|
|
mark(email, lastLimitAt=int(time.time()), disabledUntil=until, lastError=args.reason or '子邮箱已达上限邮箱')
|
|
print(json.dumps({'email':email,'disabledUntil':until,'cooldownSeconds':COOLDOWN_SECONDS}, ensure_ascii=False))
|
|
|
|
|
|
def cmd_mark_login(args):
|
|
mark(args.email.lower().strip(), lastLoginAt=int(time.time()), lastError='')
|
|
print(json.dumps({'email':args.email.lower().strip(),'ok':True}, ensure_ascii=False))
|
|
|
|
|
|
def cmd_selftest(args):
|
|
accounts=load_accounts(); state=load_state()
|
|
failures=[]
|
|
for a in accounts:
|
|
for _ in range(5):
|
|
alias=generate_alias_for(a); local=a['email'].split('@')[0]
|
|
if not (alias.startswith(local) and alias.endswith('@2925.com') and len(alias.split('@')[0])==len(local)+SUFFIX_LEN):
|
|
failures.append({'email':a['email'],'alias':alias})
|
|
st=ACCOUNTS_FILE.stat()
|
|
ok_mode=(stat.S_IMODE(st.st_mode)==0o600)
|
|
print(json.dumps({'accounts':len(accounts),'aliasFailures':failures,'secretMode':oct(stat.S_IMODE(st.st_mode)),'ok':not failures and ok_mode}, ensure_ascii=False))
|
|
raise SystemExit(0 if not failures and ok_mode else 1)
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser()
|
|
sub=p.add_subparsers(dest='cmd', required=True)
|
|
sub.add_parser('list').set_defaults(func=cmd_list)
|
|
q=sub.add_parser('pick'); q.add_argument('--preferred'); q.add_argument('--exclude', action='append'); q.add_argument('--dry-run', action='store_true'); q.set_defaults(func=cmd_pick)
|
|
l=sub.add_parser('mark-limit'); l.add_argument('email'); l.add_argument('--reason', default=''); l.set_defaults(func=cmd_limit)
|
|
m=sub.add_parser('mark-login'); m.add_argument('email'); m.set_defaults(func=cmd_mark_login)
|
|
sub.add_parser('selftest').set_defaults(func=cmd_selftest)
|
|
args=p.parse_args(); args.func(args)
|
|
if __name__=='__main__': main()
|