File: privileged.py

package info (click to toggle)
freedombox 26.2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 82,976 kB
  • sloc: python: 48,504; javascript: 1,736; xml: 481; makefile: 290; sh: 167; php: 32
file content (181 lines) | stat: -rw-r--r-- 5,335 bytes parent folder | download | duplicates (4)
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
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Configure sharing."""

import os
import pathlib
import re

import augeas

from plinth import action_utils
from plinth.actions import privileged

APACHE_CONFIGURATION = '/etc/apache2/conf-available/sharing-freedombox.conf'


@privileged
def setup():
    """Create an empty apache configuration file."""
    path = pathlib.Path(APACHE_CONFIGURATION)
    if not path.exists():
        path.touch()


def load_augeas():
    """Initialize augeas for this app's configuration file."""
    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
                        augeas.Augeas.NO_MODL_AUTOLOAD)
    aug.set('/augeas/load/Httpd/lens', 'Httpd.lns')
    aug.set('/augeas/load/Httpd/incl[last() + 1]', APACHE_CONFIGURATION)
    aug.load()

    aug.defvar('conf', '/files' + APACHE_CONFIGURATION)

    return aug


@privileged
def add(name: str, path: str, groups: list[str], is_public: bool):
    """Add a share to Apache configuration."""
    path = '"' + path.replace('"', r'\"') + '"'
    url = '/share/' + name

    if not os.path.exists(APACHE_CONFIGURATION):
        pathlib.Path(APACHE_CONFIGURATION).touch()

    aug = load_augeas()
    shares = _list(aug)
    if any([share for share in shares if share['name'] == name]):
        raise Exception('Share already present')

    aug.set('$conf/directive[last() + 1]', 'Alias')
    aug.set('$conf/directive[last()]/arg[1]', url)
    aug.set('$conf/directive[last()]/arg[2]', path)

    aug.set('$conf/Location[last() + 1]/arg', url)

    aug.set('$conf/Location[last()]/directive[last() + 1]', 'Include')
    aug.set('$conf/Location[last()]/directive[last()]/arg',
            'includes/freedombox-sharing.conf')

    if not is_public:
        aug.set('$conf/Location[last()]/directive[last() + 1]', 'Include')
        aug.set('$conf/Location[last()]/directive[last()]/arg',
                'includes/freedombox-single-sign-on.conf')

        aug.set('$conf/Location[last()]/IfModule/arg', 'mod_auth_pubtkt.c')
        aug.set('$conf/Location[last()]/IfModule/directive[1]', 'TKTAuthToken')
        for group_name in groups:
            aug.set(
                '$conf/Location[last()]/IfModule/directive[1]/arg[last() + 1]',
                group_name)
    else:
        aug.set('$conf/Location[last()]/directive[last() + 1]', 'Require')
        aug.set('$conf/Location[last()]/directive[last()]/arg[1]', 'all')
        aug.set('$conf/Location[last()]/directive[last()]/arg[2]', 'granted')

    aug.save()

    with action_utils.WebserverChange() as webserver_change:
        webserver_change.enable('sharing-freedombox')


@privileged
def remove(name: str):
    """Remove a share from Apache configuration."""
    url_to_remove = '/share/' + name

    aug = load_augeas()

    for directive in aug.match('$conf/directive'):
        if aug.get(directive) != 'Alias':
            continue

        url = aug.get(directive + '/arg[1]')
        if url == url_to_remove:
            aug.remove(directive)

    for location in aug.match('$conf/Location'):
        url = aug.get(location + '/arg')
        if url == url_to_remove:
            aug.remove(location)

    aug.save()

    with action_utils.WebserverChange() as webserver_change:
        webserver_change.enable('sharing-freedombox')


def _get_name_from_url(url):
    """Return the name of the share given the URL for it."""
    matches = re.match(r'/share/([a-z0-9\-]*)', url)
    if not matches:
        raise ValueError

    return matches[1]


def _list(aug=None):
    """List all Apache configuration shares."""
    if not aug:
        aug = load_augeas()

    shares = []

    for match in aug.match('$conf/directive'):
        if aug.get(match) != 'Alias':
            continue

        url = aug.get(match + '/arg[1]')
        path = aug.get(match + '/arg[2]')

        path = path.removesuffix('"').removeprefix('"')
        path = path.replace(r'\"', '"')
        try:
            name = _get_name_from_url(url)
            shares.append({
                'name': name,
                'path': path,
                'url': '/share/' + name
            })
        except ValueError:
            continue

    for location in aug.match('$conf/Location'):
        url = aug.get(location + '/arg')

        try:
            name = _get_name_from_url(url)
        except ValueError:
            continue

        groups = []
        for group in aug.match(location + '//directive["TKTAuthToken"]/arg'):
            groups.append(aug.get(group))

        def _is_public():
            """Must contain the line 'Require all granted'."""
            require = location + '//directive["Require"]'
            return bool(aug.match(require)) and aug.get(
                require +
                '/arg[1]') == 'all' and aug.get(require +
                                                '/arg[2]') == 'granted'

        for share in shares:
            if share['name'] == name:
                share['groups'] = groups
                share['is_public'] = _is_public()

    return shares


@privileged
def list_shares() -> list[dict[str, object]]:
    """List all Apache configuration shares and print as JSON."""
    return _list()


@privileged
def uninstall():
    """Remove apache config when app is uninstalled."""
    pathlib.Path(APACHE_CONFIGURATION).unlink(missing_ok=True)