Mercurial > public > mercurial-scm > hg-stable

comparison mercurial/hgweb/common.py @ 43077:687b865b95ad

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
formatting: byteify all mercurial/ and hgext/ string literals Done with python3.7 contrib/byteify-strings.py -i $(hg files 'set:mercurial/**.py - mercurial/thirdparty/** + hgext/**.py - hgext/fsmonitor/pywatchman/** - mercurial/__init__.py') black -l 80 -t py33 -S $(hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**" - hgext/fsmonitor/pywatchman/**') # skip-blame mass-reformatting only Differential Revision: https://phab.mercurial-scm.org/D6972
author Augie Fackler <augie@google.com>
date Sun, 06 Oct 2019 09:48:39 -0400
parents 2372284d9457
children eef9a2d67051
comparison
equal deleted inserted replaced
43076:2372284d9457 43077:687b865b95ad
40 40
41 If userlist has a single '*' member, all users are considered members. 41 If userlist has a single '*' member, all users are considered members.
42 Can be overridden by extensions to provide more complex authorization 42 Can be overridden by extensions to provide more complex authorization
43 schemes. 43 schemes.
44 """ 44 """
45 return userlist == ['*'] or username in userlist 45 return userlist == [b'*'] or username in userlist
46 46
47 47
48 def checkauthz(hgweb, req, op): 48 def checkauthz(hgweb, req, op):
49 '''Check permission for operation based on request data (including 49 '''Check permission for operation based on request data (including
50 authentication info). Return if op allowed, else raise an ErrorResponse 50 authentication info). Return if op allowed, else raise an ErrorResponse
51 exception.''' 51 exception.'''
52 52
53 user = req.remoteuser 53 user = req.remoteuser
54 54
55 deny_read = hgweb.configlist('web', 'deny_read') 55 deny_read = hgweb.configlist(b'web', b'deny_read')
56 if deny_read and (not user or ismember(hgweb.repo.ui, user, deny_read)): 56 if deny_read and (not user or ismember(hgweb.repo.ui, user, deny_read)):
57 raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized') 57 raise ErrorResponse(HTTP_UNAUTHORIZED, b'read not authorized')
58 58
59 allow_read = hgweb.configlist('web', 'allow_read') 59 allow_read = hgweb.configlist(b'web', b'allow_read')
60 if allow_read and (not ismember(hgweb.repo.ui, user, allow_read)): 60 if allow_read and (not ismember(hgweb.repo.ui, user, allow_read)):
61 raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized') 61 raise ErrorResponse(HTTP_UNAUTHORIZED, b'read not authorized')
62 62
63 if op == 'pull' and not hgweb.allowpull: 63 if op == b'pull' and not hgweb.allowpull:
64 raise ErrorResponse(HTTP_UNAUTHORIZED, 'pull not authorized') 64 raise ErrorResponse(HTTP_UNAUTHORIZED, b'pull not authorized')
65 elif op == 'pull' or op is None: # op is None for interface requests 65 elif op == b'pull' or op is None: # op is None for interface requests
66 return 66 return
67 67
68 # Allow LFS uploading via PUT requests 68 # Allow LFS uploading via PUT requests
69 if op == 'upload': 69 if op == b'upload':
70 if req.method != 'PUT': 70 if req.method != b'PUT':
71 msg = 'upload requires PUT request' 71 msg = b'upload requires PUT request'
72 raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg) 72 raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg)
73 # enforce that you can only push using POST requests 73 # enforce that you can only push using POST requests
74 elif req.method != 'POST': 74 elif req.method != b'POST':
75 msg = 'push requires POST request' 75 msg = b'push requires POST request'
76 raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg) 76 raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg)
77 77
78 # require ssl by default for pushing, auth info cannot be sniffed 78 # require ssl by default for pushing, auth info cannot be sniffed
79 # and replayed 79 # and replayed
80 if hgweb.configbool('web', 'push_ssl') and req.urlscheme != 'https': 80 if hgweb.configbool(b'web', b'push_ssl') and req.urlscheme != b'https':
81 raise ErrorResponse(HTTP_FORBIDDEN, 'ssl required') 81 raise ErrorResponse(HTTP_FORBIDDEN, b'ssl required')
82 82
83 deny = hgweb.configlist('web', 'deny_push') 83 deny = hgweb.configlist(b'web', b'deny_push')
84 if deny and (not user or ismember(hgweb.repo.ui, user, deny)): 84 if deny and (not user or ismember(hgweb.repo.ui, user, deny)):
85 raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized') 85 raise ErrorResponse(HTTP_UNAUTHORIZED, b'push not authorized')
86 86
87 allow = hgweb.configlist('web', 'allow-push') 87 allow = hgweb.configlist(b'web', b'allow-push')
88 if not (allow and ismember(hgweb.repo.ui, user, allow)): 88 if not (allow and ismember(hgweb.repo.ui, user, allow)):
89 raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized') 89 raise ErrorResponse(HTTP_UNAUTHORIZED, b'push not authorized')
90 90
91 91
92 # Hooks for hgweb permission checks; extensions can add hooks here. 92 # Hooks for hgweb permission checks; extensions can add hooks here.
93 # Each hook is invoked like this: hook(hgweb, request, operation), 93 # Each hook is invoked like this: hook(hgweb, request, operation),
94 # where operation is either read, pull, push or upload. Hooks should either 94 # where operation is either read, pull, push or upload. Hooks should either
126 self.continued = False 126 self.continued = False
127 127
128 def read(self, amt=-1): 128 def read(self, amt=-1):
129 if not self.continued: 129 if not self.continued:
130 self.continued = True 130 self.continued = True
131 self._write('HTTP/1.1 100 Continue\r\n\r\n') 131 self._write(b'HTTP/1.1 100 Continue\r\n\r\n')
132 return self.f.read(amt) 132 return self.f.read(amt)
133 133
134 def __getattr__(self, attr): 134 def __getattr__(self, attr):
135 if attr in ('close', 'readline', 'readlines', '__iter__'): 135 if attr in (b'close', b'readline', b'readlines', b'__iter__'):
136 return getattr(self.f, attr) 136 return getattr(self.f, attr)
137 raise AttributeError 137 raise AttributeError
138 138
139 139
140 def _statusmessage(code): 140 def _statusmessage(code):
143 responses.get(code, (r'Error', r'Unknown error'))[0] 143 responses.get(code, (r'Error', r'Unknown error'))[0]
144 ) 144 )
145 145
146 146
147 def statusmessage(code, message=None): 147 def statusmessage(code, message=None):
148 return '%d %s' % (code, message or _statusmessage(code)) 148 return b'%d %s' % (code, message or _statusmessage(code))
149 149
150 150
151 def get_stat(spath, fn): 151 def get_stat(spath, fn):
152 """stat fn if it exists, spath otherwise""" 152 """stat fn if it exists, spath otherwise"""
153 cl_path = os.path.join(spath, fn) 153 cl_path = os.path.join(spath, fn)
156 else: 156 else:
157 return os.stat(spath) 157 return os.stat(spath)
158 158
159 159
160 def get_mtime(spath): 160 def get_mtime(spath):
161 return get_stat(spath, "00changelog.i")[stat.ST_MTIME] 161 return get_stat(spath, b"00changelog.i")[stat.ST_MTIME]
162 162
163 163
164 def ispathsafe(path): 164 def ispathsafe(path):
165 """Determine if a path is safe to use for filesystem access.""" 165 """Determine if a path is safe to use for filesystem access."""
166 parts = path.split('/') 166 parts = path.split(b'/')
167 for part in parts: 167 for part in parts:
168 if ( 168 if (
169 part in ('', pycompat.oscurdir, pycompat.ospardir) 169 part in (b'', pycompat.oscurdir, pycompat.ospardir)
170 or pycompat.ossep in part 170 or pycompat.ossep in part
171 or pycompat.osaltsep is not None 171 or pycompat.osaltsep is not None
172 and pycompat.osaltsep in part 172 and pycompat.osaltsep in part
173 ): 173 ):
174 return False 174 return False
186 186
187 """ 187 """
188 if not ispathsafe(fname): 188 if not ispathsafe(fname):
189 return 189 return
190 190
191 fpath = os.path.join(*fname.split('/')) 191 fpath = os.path.join(*fname.split(b'/'))
192 if isinstance(directory, str): 192 if isinstance(directory, str):
193 directory = [directory] 193 directory = [directory]
194 for d in directory: 194 for d in directory:
195 path = os.path.join(d, fpath) 195 path = os.path.join(d, fpath)
196 if os.path.exists(path): 196 if os.path.exists(path):
198 try: 198 try:
199 os.stat(path) 199 os.stat(path)
200 ct = pycompat.sysbytes( 200 ct = pycompat.sysbytes(
201 mimetypes.guess_type(pycompat.fsdecode(path))[0] or r"text/plain" 201 mimetypes.guess_type(pycompat.fsdecode(path))[0] or r"text/plain"
202 ) 202 )
203 with open(path, 'rb') as fh: 203 with open(path, b'rb') as fh:
204 data = fh.read() 204 data = fh.read()
205 205
206 res.headers['Content-Type'] = ct 206 res.headers[b'Content-Type'] = ct
207 res.setbodybytes(data) 207 res.setbodybytes(data)
208 return res 208 return res
209 except TypeError: 209 except TypeError:
210 raise ErrorResponse(HTTP_SERVER_ERROR, 'illegal filename') 210 raise ErrorResponse(HTTP_SERVER_ERROR, b'illegal filename')
211 except OSError as err: 211 except OSError as err:
212 if err.errno == errno.ENOENT: 212 if err.errno == errno.ENOENT:
213 raise ErrorResponse(HTTP_NOT_FOUND) 213 raise ErrorResponse(HTTP_NOT_FOUND)
214 else: 214 else:
215 raise ErrorResponse( 215 raise ErrorResponse(
239 239
240 web.contact is the primary source, but if that is not set, try 240 web.contact is the primary source, but if that is not set, try
241 ui.username or $EMAIL as a fallback to display something useful. 241 ui.username or $EMAIL as a fallback to display something useful.
242 """ 242 """
243 return ( 243 return (
244 config("web", "contact") 244 config(b"web", b"contact")
245 or config("ui", "username") 245 or config(b"ui", b"username")
246 or encoding.environ.get("EMAIL") 246 or encoding.environ.get(b"EMAIL")
247 or "" 247 or b""
248 ) 248 )
249 249
250 250
251 def cspvalues(ui): 251 def cspvalues(ui):
252 """Obtain the Content-Security-Policy header and nonce value. 252 """Obtain the Content-Security-Policy header and nonce value.
273 # We can move it back once we no longer need Python <= 2.7.12 support. 273 # We can move it back once we no longer need Python <= 2.7.12 support.
274 import uuid 274 import uuid
275 275
276 # Don't allow untrusted CSP setting since it be disable protections 276 # Don't allow untrusted CSP setting since it be disable protections
277 # from a trusted/global source. 277 # from a trusted/global source.
278 csp = ui.config('web', 'csp', untrusted=False) 278 csp = ui.config(b'web', b'csp', untrusted=False)
279 nonce = None 279 nonce = None
280 280
281 if csp and '%nonce%' in csp: 281 if csp and b'%nonce%' in csp:
282 nonce = base64.urlsafe_b64encode(uuid.uuid4().bytes).rstrip('=') 282 nonce = base64.urlsafe_b64encode(uuid.uuid4().bytes).rstrip(b'=')
283 csp = csp.replace('%nonce%', nonce) 283 csp = csp.replace(b'%nonce%', nonce)
284 284
285 return csp, nonce 285 return csp, nonce