File: common.py

package info (click to toggle)
ldh-client 0.0.9-1
  • links: PTS, VCS
  • area: main
  • in suites: experimental
  • size: 228 kB
  • sloc: python: 245; makefile: 17; sh: 5
file content (152 lines) | stat: -rwxr-xr-x 3,565 bytes parent folder | download
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
#!/usr/bin/env python3
# Copyright 2017-2020 Purism SPC
# SPDX-License-Identifier: AGPL-3.0-or-later

import ldh_client
import re

from getpass import getpass

# define strings that use defaults
ADDRESS_PROMPT = "Enter your " + str(ldh_client.DEFAULT["HOST_LABEL"]) + " address: "
PASSPHRASE_PROMPT = "Enter your passphrase: "


class LdhCredential(object):
    """User credentials for an LDH account."""

    def __init__(self, user, host, passphrase):
        self.user = user
        self.host = host
        self.passphrase = passphrase

    def __str__(self):
        return self.address + " : '" + self.passphrase + "'"

    @property
    def address(self):
        return self.user + "@" + self.host


class LdhError(Exception):
    """Base class for LDH-related errors."""

    def __init__(self, message):
        self.message = message


def prompt_for_credentials():
    """Prompts the user until they enter valid credentials. Returns a valid LdhCredential."""

    valid_address = False

    while not valid_address:
        address = input(ADDRESS_PROMPT)
        regex = r"^[A-Za-z][A-Za-z0-9]*@[A-Za-z0-9]+(\.[A-Za-z0-9]+)+$"
        if not re.match(regex, address):
            print(address + " is not a valid email address.")
        else:
            valid_address = True

    (user, host) = address.split("@")
    passphrase = getpass(PASSPHRASE_PROMPT)

    return LdhCredential(user, host, passphrase)


def get_accounts_from_goa():

    try:
        import gi
        gi.require_version('Goa', '1.0')
    except ValueError:
        raise ImportError('GOA library not available')

    from gi.repository import Goa

    client = Goa.Client.new_sync()
    return client.get_accounts()


def count_goa_credentials():

    count = 0

    try:
        accounts = get_accounts_from_goa()
        for objprx in accounts:
            accprx = objprx.get_account()
            if accprx.props.provider_type == 'librem_one':
                count += 1
    except Exception:
        count = -1

    return count


def get_single_goa_credential():
    """Query GOA for valid credentials."""

    user = None
    host = None
    passphrase = None

    accounts = get_accounts_from_goa()
    count = 0
    for objprx in accounts:
        accprx = objprx.get_account()
        if accprx.props.provider_type == 'librem_one':
            count += 1
            address = accprx.props.identity
            (user, host) = address.split("@")
            pbobj = objprx.get_password_based()
            passphrase = pbobj.call_get_password_sync('password')

    if count == 0:
        raise ValueError("No GOA credentials found.")
    elif count > 1:
        raise ValueError("Too many GOA credentials found.")

    return LdhCredential(user, host, passphrase)


def get_rclone_version():
    """Extract version number from rclone, if it is available."""

    try:

        import sh
        from sh import rclone
        from io import StringIO

        buffer = StringIO()

        try:
            rclone("--version", _out=buffer)
        except sh.ErrorReturnCode_252:
            pass

        return buffer.getvalue()
    except Exception:
        return "None"


def get_yad_version():
    """Extract version number from yad, if it is available."""

    try:

        import sh
        from sh import yad
        from io import StringIO

        buffer = StringIO()

        try:
            yad("--version", _out=buffer)
        except sh.ErrorReturnCode_252:
            pass

        return buffer.getvalue()
    except Exception:
        return "None"